[tests][OK] Using a dummy_domain, adding it and listing it's routes
This commit is contained in:
parent
f5ebabbcd4
commit
40547ddf30
|
@ -1,43 +1,73 @@
|
|||
#!/usr/bin/env python3
|
||||
import logging
|
||||
import functools
|
||||
import re
|
||||
import os
|
||||
import subprocess
|
||||
import importlib
|
||||
import tempfile
|
||||
import click
|
||||
from unittest.mock import patch
|
||||
from typing import Dict, Tuple
|
||||
import pytest
|
||||
from uuid import uuid1
|
||||
from click.testing import CliRunner
|
||||
from halfapi import __version__
|
||||
from halfapi.cli import cli
|
||||
from halfapi.cli.init import format_halfapi_etc
|
||||
Cli = cli.cli
|
||||
from halfapi.cli.cli import cli
|
||||
from halfapi.cli.init import init, format_halfapi_etc
|
||||
from halfapi.cli.domain import domain, create_domain
|
||||
logger = logging.getLogger('halfapitest')
|
||||
|
||||
PROJNAME = os.environ.get('PROJ','tmp_api')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner():
|
||||
return CliRunner()
|
||||
|
||||
def halfapicli(runner):
|
||||
return lambda *args: runner.invoke(Cli, args)
|
||||
|
||||
@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 dropdb():
|
||||
p = subprocess.Popen(['dropdb', f'halfapi_{PROJNAME}'])
|
||||
p.wait()
|
||||
yield
|
||||
def halfapicli(cli_runner):
|
||||
def caller(*args):
|
||||
return cli_runner.invoke(cli, ' '.join(args))
|
||||
|
||||
p = subprocess.Popen(['dropdb', f'halfapi_{PROJNAME}'])
|
||||
p.wait()
|
||||
yield caller
|
||||
|
||||
@pytest.fixture
|
||||
def createdb():
|
||||
p = subprocess.Popen(['createdb', f'halfapi_{PROJNAME}'])
|
||||
p.wait()
|
||||
return
|
||||
def halfapi_conf_dir():
|
||||
return confdir('HALFAPI_CONF_DIR')
|
||||
|
||||
|
||||
|
||||
def confdir(dirname):
|
||||
d = os.environ.get(dirname)
|
||||
|
@ -52,10 +82,6 @@ def confdir(dirname):
|
|||
def halform_conf_dir():
|
||||
return confdir('HALFORM_CONF_DIR')
|
||||
|
||||
@pytest.fixture
|
||||
def halfapi_conf_dir():
|
||||
return confdir('HALFAPI_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]] = {}
|
||||
|
@ -106,28 +132,32 @@ def pytest_runtest_setup(item):
|
|||
pytest.xfail("previous test failed ({})".format(test_name))
|
||||
|
||||
@pytest.fixture
|
||||
def project_runner(runner, halfapi_conf_dir):
|
||||
global Cli
|
||||
env = {
|
||||
'HALFAPI_CONF_DIR': halfapi_conf_dir
|
||||
}
|
||||
def project_runner(runner, halfapicli, halfapi_conf_dir):
|
||||
with runner.isolated_filesystem():
|
||||
res = runner.invoke(Cli, ['init', PROJNAME],
|
||||
env=env,
|
||||
catch_exceptions=True)
|
||||
assert res.exit_code == 0
|
||||
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 f:
|
||||
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()))
|
||||
f.write(PROJ_CONFIG)
|
||||
halfapi_etc.write(PROJ_CONFIG)
|
||||
|
||||
importlib.reload(cli)
|
||||
Cli = cli.cli
|
||||
yield halfapicli(runner)
|
||||
|
||||
###
|
||||
# add dummy domain
|
||||
###
|
||||
create_domain('dummy_domain', '.dummy_domain')
|
||||
###
|
||||
|
||||
yield halfapicli
|
||||
|
|
|
@ -0,0 +1 @@
|
|||
public = lambda *args: True
|
|
@ -1,3 +1,5 @@
|
|||
from starlette.responses import PlainTextResponse
|
||||
|
||||
ROUTES={
|
||||
'': {
|
||||
'GET': [{'acl':None, 'in':None}]
|
||||
|
@ -10,3 +12,11 @@ ROUTES={
|
|||
}
|
||||
|
||||
}
|
||||
|
||||
async def get(request, *args, **kwargs):
|
||||
"""
|
||||
responses:
|
||||
200:
|
||||
description: dummy abc.alphabet route
|
||||
"""
|
||||
return PlainTextResponse('True')
|
||||
|
|
|
@ -18,7 +18,7 @@ PROJNAME = os.environ.get('PROJ','tmp_api')
|
|||
|
||||
@pytest.mark.incremental
|
||||
class TestCli():
|
||||
def test_options(self, runner, dropdb, createdb):
|
||||
def test_options(self, runner):
|
||||
# Wrong command
|
||||
with runner.isolated_filesystem():
|
||||
r = runner.invoke(Cli, ['foobar'])
|
||||
|
@ -38,7 +38,7 @@ class TestCli():
|
|||
assert r.exit_code == 0
|
||||
|
||||
|
||||
def test_init_project_fail(self, runner, dropdb):
|
||||
def test_init_project_fail(self, runner):
|
||||
# Missing argument (project)
|
||||
testproject = 'testproject'
|
||||
r = runner.invoke(Cli, ['init'])
|
||||
|
|
|
@ -15,15 +15,21 @@ PROJNAME = os.environ.get('PROJ','tmp_api')
|
|||
@pytest.mark.incremental
|
||||
class TestCliProj():
|
||||
def test_cmds(self, project_runner):
|
||||
print(os.getcwd())
|
||||
assert project_runner('--help').exit_code == 0
|
||||
#assert project_runner('run', '--help').exit_code == 0
|
||||
#assert project_runner('domain', '--help').exit_code == 0
|
||||
|
||||
|
||||
def test_domain_commands(self, project_runner):
|
||||
r = project_runner('domain')
|
||||
assert r.exit_code == 0
|
||||
|
||||
@pytest.mark.skip
|
||||
def test_config_commands(self, project_runner):
|
||||
assert project_runner('config') == 0
|
||||
try:
|
||||
r = project_runner('config')
|
||||
assert r.exit_code == 0
|
||||
except AssertionError as exc:
|
||||
subprocess.call(['tree', '-a'])
|
||||
raise exc
|
||||
|
||||
|
||||
def test_domain_commands(self, subproc):
|
||||
assert project_runner('domain') == 0
|
||||
|
|
|
@ -1,12 +1,11 @@
|
|||
#!/usr/bin/env python3
|
||||
from starlette.routing import Route
|
||||
from halfapi.lib.domain import VERBS, gen_router_routes
|
||||
import importlib
|
||||
from halfapi.lib.domain import VERBS, gen_domain_routes, gen_router_routes
|
||||
|
||||
from halfapi.lib.routes import gen_starlette_routes
|
||||
|
||||
def test_gen_router_routes():
|
||||
from .dummy_domain import routers
|
||||
for path, d_route in gen_router_routes(routers):
|
||||
for path, d_route in gen_router_routes(routers, ['dummy_domain']):
|
||||
assert isinstance(path, str)
|
||||
for verb in VERBS:
|
||||
if verb not in d_route.keys():
|
||||
|
@ -15,12 +14,12 @@ def test_gen_router_routes():
|
|||
print(f'[{verb}] {path} {route["fct"]}')
|
||||
assert len(route['params']) > 0
|
||||
assert hasattr(route['fct'], '__call__')
|
||||
if hasattr('fqtn', route):
|
||||
if 'fqtn' in route:
|
||||
assert isinstance(route['fqtn'], str)
|
||||
|
||||
|
||||
def test_gen_starlette_routes():
|
||||
def test_gen_domain_routes():
|
||||
from . import dummy_domain
|
||||
for route in gen_starlette_routes(dummy_domain):
|
||||
assert isinstance(route, Route)
|
||||
|
||||
for route in gen_domain_routes(
|
||||
'dummy_domain', dummy_domain):
|
||||
assert isinstance(route, dict)
|
||||
|
|
|
@ -0,0 +1,10 @@
|
|||
from starlette.routing import Route
|
||||
from halfapi.lib.routes import gen_starlette_routes
|
||||
|
||||
def test_gen_starlette_routes():
|
||||
from . import dummy_domain
|
||||
for path, route in gen_starlette_routes({
|
||||
'dummy_domain': dummy_domain }):
|
||||
|
||||
assert isinstance(path, str)
|
||||
assert isinstance(route, Route)
|
|
@ -0,0 +1,30 @@
|
|||
import subprocess
|
||||
from pprint import pprint
|
||||
from starlette.testclient import TestClient
|
||||
from starlette.authentication import (
|
||||
AuthenticationBackend, AuthenticationError, BaseUser, AuthCredentials,
|
||||
UnauthenticatedUser)
|
||||
|
||||
from halfapi.app import application
|
||||
from halfapi.lib.schemas import schema_dict_dom
|
||||
|
||||
def test_schemas_dict_dom():
|
||||
from . import dummy_domain
|
||||
schema = schema_dict_dom({
|
||||
'dummy_domain':dummy_domain})
|
||||
assert isinstance(schema, dict)
|
||||
|
||||
|
||||
def test_get_api_routes(project_runner):
|
||||
c = TestClient(application)
|
||||
r = c.get('/')
|
||||
d_r = r.json()
|
||||
try:
|
||||
assert isinstance(d_r, dict)
|
||||
assert 'paths' in d_r
|
||||
assert '/' in d_r['paths']
|
||||
assert '/dummy_domain/abc/alphabet' in d_r['paths']
|
||||
except AssertionError as exc:
|
||||
pprint(d_r)
|
||||
raise exc
|
||||
|
Loading…
Reference in New Issue