304 lines
8.0 KiB
Python
304 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
import logging
|
|
import functools
|
|
import re
|
|
import os
|
|
import subprocess
|
|
import importlib
|
|
import tempfile
|
|
from typing import Dict, Tuple
|
|
from uuid import uuid1, uuid4, UUID
|
|
import click
|
|
from click.testing import CliRunner
|
|
import jwt
|
|
import sys
|
|
from unittest.mock import patch
|
|
import pytest
|
|
from starlette.applications import Starlette
|
|
from starlette.responses import PlainTextResponse
|
|
from starlette.middleware.authentication import AuthenticationMiddleware
|
|
from starlette.testclient import TestClient
|
|
from halfapi import __version__
|
|
from halfapi.cli.cli import cli
|
|
from halfapi.cli.init import init, format_halfapi_etc
|
|
from halfapi.cli.domain import domain, create_domain
|
|
from halfapi.lib.responses import ORJSONResponse
|
|
from halfapi.lib.jwt_middleware import JWTAuthenticationBackend
|
|
logger = logging.getLogger('halfapitest')
|
|
|
|
PROJNAME = os.environ.get('PROJ','tmp_api')
|
|
|
|
os.environ['HALFAPI_SECRET'] = 'dummysecret'
|
|
SECRET = 'dummysecret'
|
|
|
|
from halfapi.lib.jwt_middleware import (
|
|
JWTUser, JWTAuthenticationBackend,
|
|
JWTWebSocketAuthenticationBackend)
|
|
|
|
@pytest.fixture
|
|
def token_builder():
|
|
yield jwt.encode({
|
|
'name':'xxx',
|
|
'user_id': str(uuid4())},
|
|
key=SECRET
|
|
)
|
|
|
|
@pytest.fixture
|
|
def token_debug_false_builder():
|
|
yield jwt.encode({
|
|
'name':'xxx',
|
|
'user_id': str(uuid4()),
|
|
'debug': False},
|
|
key=SECRET
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def token_debug_true_builder():
|
|
yield jwt.encode({
|
|
'name':'xxx',
|
|
'user_id': str(uuid4()),
|
|
'debug': True},
|
|
key=SECRET
|
|
)
|
|
|
|
@pytest.fixture
|
|
def runner():
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def cli_runner():
|
|
"""Yield a click.testing.CliRunner to invoke the CLI."""
|
|
class_ = click.testing.CliRunner
|
|
|
|
def invoke_wrapper(f):
|
|
"""Augment CliRunner.invoke to emit its output to stdout.
|
|
|
|
This enables pytest to show the output in its logs on test
|
|
failures.
|
|
|
|
"""
|
|
@functools.wraps(f)
|
|
def wrapper(*args, **kwargs):
|
|
echo = kwargs.pop('echo', False)
|
|
result = f(*args, **kwargs)
|
|
|
|
if echo is True:
|
|
sys.stdout.write(result.output)
|
|
|
|
return result
|
|
|
|
return wrapper
|
|
|
|
class_.invoke = invoke_wrapper(class_.invoke)
|
|
cli_runner_ = class_()
|
|
|
|
yield cli_runner_
|
|
|
|
|
|
@pytest.fixture
|
|
def halfapicli(cli_runner):
|
|
def caller(*args):
|
|
return cli_runner.invoke(cli, ' '.join(args))
|
|
|
|
yield caller
|
|
|
|
@pytest.fixture
|
|
def halfapi_conf_dir():
|
|
return confdir('HALFAPI_CONF_DIR')
|
|
|
|
|
|
|
|
def confdir(dirname):
|
|
d = os.environ.get(dirname)
|
|
if not d:
|
|
os.environ[dirname] = tempfile.mkdtemp(prefix='halfapi_')
|
|
return os.environ.get(dirname)
|
|
if not os.path.isdir(d):
|
|
os.mkdir(d)
|
|
return d
|
|
|
|
@pytest.fixture
|
|
def halform_conf_dir():
|
|
return confdir('HALFORM_CONF_DIR')
|
|
|
|
# store history of failures per test class name and per index in parametrize (if
|
|
# parametrize used)
|
|
_test_failed_incremental: Dict[str, Dict[Tuple[int, ...], str]] = {}
|
|
|
|
|
|
def pytest_runtest_makereport(item, call):
|
|
if "incremental" in item.keywords:
|
|
# incremental marker is used
|
|
if call.excinfo is not None:
|
|
# the test has failed
|
|
# retrieve the class name of the test
|
|
cls_name = str(item.cls)
|
|
# retrieve the index of the test (if parametrize is used in
|
|
# combination with incremental)
|
|
parametrize_index = (
|
|
tuple(item.callspec.indices.values())
|
|
if hasattr(item, "callspec")
|
|
else ()
|
|
)
|
|
# retrieve the name of the test function
|
|
test_name = item.originalname or item.name
|
|
# store in _test_failed_incremental the original name of the failed
|
|
# test
|
|
_test_failed_incremental.setdefault(cls_name, {}).setdefault(
|
|
parametrize_index, test_name
|
|
)
|
|
|
|
|
|
def pytest_runtest_setup(item):
|
|
if "incremental" in item.keywords:
|
|
# retrieve the class name of the test
|
|
cls_name = str(item.cls)
|
|
# check if a previous test has failed for this class
|
|
if cls_name in _test_failed_incremental:
|
|
# retrieve the index of the test (if parametrize is used in
|
|
# combination with incremental)
|
|
parametrize_index = (
|
|
tuple(item.callspec.indices.values())
|
|
if hasattr(item, "callspec")
|
|
else ()
|
|
)
|
|
# retrieve the name of the first test function to fail for this
|
|
# class name and index
|
|
test_name = _test_failed_incremental[cls_name].get(parametrize_index, None)
|
|
# if name found, test has failed for the combination of class name &
|
|
# test name
|
|
if test_name is not None:
|
|
pytest.xfail("previous test failed ({})".format(test_name))
|
|
|
|
@pytest.fixture
|
|
def project_runner(runner, halfapicli, halfapi_conf_dir):
|
|
with runner.isolated_filesystem():
|
|
res = halfapicli('init', PROJNAME)
|
|
|
|
try:
|
|
os.chdir(PROJNAME)
|
|
except FileNotFoundError as exc:
|
|
subprocess.call('tree')
|
|
raise exc
|
|
|
|
|
|
secret = tempfile.mkstemp()
|
|
SECRET_PATH = secret[1]
|
|
with open(SECRET_PATH, 'w') as f:
|
|
f.write(str(uuid1()))
|
|
|
|
with open(os.path.join(halfapi_conf_dir, PROJNAME), 'w') as halfapi_etc:
|
|
PROJ_CONFIG = re.sub('secret = .*', f'secret = {SECRET_PATH}',
|
|
format_halfapi_etc(PROJNAME, os.getcwd()))
|
|
halfapi_etc.write(PROJ_CONFIG)
|
|
|
|
|
|
###
|
|
# add dummy domain
|
|
###
|
|
create_domain('tests', '.dummy_domain.routers')
|
|
###
|
|
|
|
yield halfapicli
|
|
|
|
@pytest.fixture
|
|
def dummy_app():
|
|
app = Starlette()
|
|
app.add_route('/',
|
|
lambda request, *args, **kwargs: PlainTextResponse('Hello test!'))
|
|
app.add_middleware(
|
|
AuthenticationMiddleware,
|
|
backend=JWTAuthenticationBackend(secret_key='dummysecret')
|
|
)
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def dummy_debug_app():
|
|
app = Starlette(debug=True)
|
|
app.add_route('/',
|
|
lambda request, *args, **kwargs: PlainTextResponse('Hello test!'))
|
|
app.add_middleware(
|
|
AuthenticationMiddleware,
|
|
backend=JWTAuthenticationBackend(secret_key='dummysecret')
|
|
)
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def test_client(dummy_app):
|
|
return TestClient(dummy_app)
|
|
|
|
@pytest.fixture
|
|
def create_route():
|
|
def wrapped(domain_path, method, path):
|
|
stack = [domain_path, *path.split('/')[1:]]
|
|
for i in range(len(stack)):
|
|
if len(stack[i]) == 0:
|
|
continue
|
|
|
|
path = os.path.join(*stack[0:i+1])
|
|
if os.path.isdir(os.path.join(path)):
|
|
continue
|
|
os.mkdir(path)
|
|
init_path = os.path.join(*stack, '__init__.py')
|
|
with open(init_path, 'a+') as f:
|
|
f.write(f'\ndef {method}():\n raise NotImplementedError')
|
|
|
|
return wrapped
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
def dummy_project():
|
|
sys.path.insert(0, './tests')
|
|
halfapi_config = tempfile.mktemp()
|
|
halfapi_secret = tempfile.mktemp()
|
|
domain = 'dummy_domain'
|
|
|
|
with open(halfapi_config, 'w') as f:
|
|
f.writelines([
|
|
'[project]\n',
|
|
'name = lirmm_api\n',
|
|
'halfapi_version = 0.5.0\n',
|
|
f'secret = {halfapi_secret}\n',
|
|
'port = 3050\n',
|
|
'loglevel = debug\n',
|
|
'[domains]\n',
|
|
f'{domain}= .routers'
|
|
])
|
|
|
|
with open(halfapi_secret, 'w') as f:
|
|
f.write('turlututu')
|
|
|
|
return (halfapi_config, 'dummy_domain', 'routers')
|
|
|
|
@pytest.fixture
|
|
def routers():
|
|
sys.path.insert(0, './tests')
|
|
|
|
from dummy_domain import routers
|
|
return routers
|
|
|
|
|
|
@pytest.fixture
|
|
def application_debug():
|
|
from halfapi.app import HalfAPI
|
|
return HalfAPI({
|
|
'SECRET':'turlututu',
|
|
'PRODUCTION':False
|
|
}).application
|
|
|
|
|
|
@pytest.fixture
|
|
def application_domain(routers):
|
|
from halfapi.app import HalfAPI
|
|
return HalfAPI({
|
|
'SECRET':'turlututu',
|
|
'PRODUCTION':True,
|
|
'DOMAINS':{'dummy_domain':routers}
|
|
}).application
|
|
|
|
|