Compare commits

...

13 Commits

Author SHA1 Message Date
Maxime Alves LIRMM@home ac935db6d6 [tests] remove dummy-domain from dependencies 2021-06-17 18:52:49 +02:00
Maxime Alves LIRMM@home 4d50363c9b [tests] update tests for 0.5.3 2021-06-17 18:52:18 +02:00
Maxime Alves LIRMM@home 6181592692 [lib.*] Refactor libs 2021-06-17 18:52:09 +02:00
Maxime Alves LIRMM@home ed7485a8a1 [app] Use HalfAPI class to be able to use custom configuration
à
2021-06-17 18:15:10 +02:00
Maxime Alves LIRMM fa1ca6bf9d [wip] tests dummy_domain 2021-06-16 15:34:25 +02:00
Maxime Alves LIRMM 86e8dd3465 [0.5.3] ajout de la config actuelle dans les arguments des routes 2021-06-15 18:12:13 +02:00
Maxime Alves LIRMM aa7ec62c7a [lib.jwtMw] verify signature even if halfapi is in DEBUG mode 2021-06-15 11:16:23 +02:00
Maxime Alves LIRMM e208728d7e [lib.acl] args_check doesn't check required/optional arguments if "args" is not specified in request, if the target function is not async 2021-06-15 11:01:50 +02:00
Maxime Alves LIRMM aa4c309778 [lib.domain] SUBROUTER can be a path parameter if including ":" 2021-06-15 09:45:37 +02:00
Maxime Alves LIRMM@home 138420461d [gitignore] *.swp 2021-06-15 07:24:32 +02:00
Maxime Alves LIRMM@home 0c1e2849ba [tests] test get route with dummy projects 2021-06-15 07:24:14 +02:00
Maxime Alves LIRMM@home 7227e2d7f1 [lib.domain] handle modules without ROUTES attribute 2021-06-14 17:18:47 +02:00
Maxime Alves LIRMM@home 78c75cd60e [tests] add dummy_project_router tests for path-based routers (without ROUTES variable) 2021-06-14 16:34:58 +02:00
27 changed files with 564 additions and 211 deletions

3
.gitignore vendored
View File

@ -138,3 +138,6 @@ cython_debug/
domains/
.vscode
# Vim swap files
*.swp

View File

@ -8,7 +8,6 @@ pytest = "*"
requests = "*"
pytest-asyncio = "*"
pylint = "*"
dummy-domain = {path = "./tests"}
[packages]
click = ">=7.1,<8"

View File

@ -1,5 +1,5 @@
#!/usr/bin/env python3
__version__ = '0.5.2'
__version__ = '0.5.3'
def version():
return f'HalfAPI version:{__version__}'

View File

@ -23,7 +23,6 @@ from timing_asgi import TimingMiddleware
from timing_asgi.integrations import StarletteScopeToName
# module libraries
from halfapi.conf import config, SECRET, PRODUCTION, DOMAINSDICT
from .lib.domain_middleware import DomainMiddleware
from .lib.timing import HTimingClient
@ -34,61 +33,86 @@ from halfapi.lib.responses import (ORJSONResponse, UnauthorizedResponse,
NotFoundResponse, InternalServerErrorResponse, NotImplementedResponse)
from halfapi.lib.routes import gen_starlette_routes, debug_routes
from halfapi.lib.schemas import get_api_routes, schema_json, get_acls
from halfapi.lib.schemas import get_api_routes, get_api_domain_routes, schema_json, get_acls
logger = logging.getLogger('uvicorn.asgi')
routes = [ Route('/', get_api_routes) ]
class HalfAPI:
def __init__(self, config=None):
if config:
SECRET = config.get('SECRET')
PRODUCTION = config.get('PRODUCTION')
DOMAINS = config.get('DOMAINS', {})
CONFIG = config.get('CONFIG', {
'domains': DOMAINS
})
else:
from halfapi.conf import CONFIG, SECRET, PRODUCTION, DOMAINS
routes += [
Route('/halfapi/schema', schema_json),
Route('/halfapi/acls', get_acls)
]
routes += Route('/halfapi/current_user', lambda request, *args, **kwargs:
ORJSONResponse({'user':request.user.json})
if SECRET and not isinstance(request.user, UnauthenticatedUser)
else ORJSONResponse({'user': None})),
routes = [ Route('/', get_api_routes) ]
if not PRODUCTION:
for route in debug_routes():
routes.append( route )
routes += [
Route('/halfapi/schema', schema_json),
Route('/halfapi/acls', get_acls)
]
routes += Route('/halfapi/current_user', lambda request, *args, **kwargs:
ORJSONResponse({'user':request.user.json})
if SECRET and not isinstance(request.user, UnauthenticatedUser)
else ORJSONResponse({'user': None})),
if DOMAINSDICT:
for route in gen_starlette_routes(DOMAINSDICT()):
routes.append(route)
if not PRODUCTION:
for route in debug_routes():
routes.append( route )
application = Starlette(
debug=not PRODUCTION,
routes=routes,
exception_handlers={
401: UnauthorizedResponse,
404: NotFoundResponse,
500: InternalServerErrorResponse,
501: NotImplementedResponse
}
)
if DOMAINS:
for route in gen_starlette_routes(DOMAINS):
routes.append(route)
if DOMAINSDICT:
application.add_middleware(
DomainMiddleware,
config=config
)
for domain in DOMAINS:
routes.append(
Route(
f'/{domain}',
get_api_domain_routes(domain)
)
)
if SECRET:
application.add_middleware(
AuthenticationMiddleware,
backend=JWTAuthenticationBackend(secret_key=SECRET)
)
if not PRODUCTION:
application.add_middleware(
TimingMiddleware,
client=HTimingClient(),
metric_namer=StarletteScopeToName(prefix="halfapi",
starlette_app=application)
)
self.application = Starlette(
debug=not PRODUCTION,
routes=routes,
exception_handlers={
401: UnauthorizedResponse,
404: NotFoundResponse,
500: InternalServerErrorResponse,
501: NotImplementedResponse
}
)
self.application.add_middleware(
DomainMiddleware,
config=CONFIG
)
if SECRET:
self.application.add_middleware(
AuthenticationMiddleware,
backend=JWTAuthenticationBackend(secret_key=SECRET)
)
if not PRODUCTION:
self.application.add_middleware(
TimingMiddleware,
client=HTimingClient(),
metric_namer=StarletteScopeToName(prefix="halfapi",
starlette_app=self.application)
)
application = HalfAPI().application

View File

@ -47,13 +47,15 @@ logger = logging.getLogger('halfapi')
PROJECT_NAME = os.path.basename(os.getcwd())
DOMAINSDICT = lambda: {}
DOMAINS = {}
PRODUCTION = False
LOGLEVEL = 'info'
HOST = '127.0.0.1'
PORT = '3000'
SECRET = ''
CONF_FILE = os.environ.get('HALFAPI_CONF_FILE', '.halfapi/config')
IS_PROJECT = os.path.isfile('.halfapi/config')
is_project = lambda: os.path.isfile(CONF_FILE)
@ -77,7 +79,7 @@ HALFAPI_ETC_FILE=os.path.join(
HALFAPI_DOT_FILE=os.path.join(
os.getcwd(), '.halfapi', 'config')
HALFAPI_CONFIG_FILES = [ HALFAPI_ETC_FILE, HALFAPI_DOT_FILE ]
HALFAPI_CONFIG_FILES = [ CONF_FILE, HALFAPI_DOT_FILE ]
def conf_files():
return [
@ -114,8 +116,11 @@ def read_config():
if IS_PROJECT:
read_config()
CONFIG = {}
IS_PROJECT = False
if is_project():
IS_PROJECT = True
CONFIG = read_config()
PROJECT_NAME = config.get('project', 'name', fallback=PROJECT_NAME)
@ -123,12 +128,14 @@ if IS_PROJECT:
raise Exception('Need a project name as argument')
DOMAINSDICT = lambda: d_domains(config)
DOMAINS = DOMAINSDICT()
HOST = config.get('project', 'host')
PORT = config.getint('project', 'port')
try:
with open(config.get('project', 'secret')) as secret_file:
SECRET = secret_file.read()
SECRET = secret_file.read().strip()
CONFIG['secret'] = SECRET.strip()
# Set the secret so we can use it in domains
os.environ['HALFAPI_SECRET'] = SECRET
except FileNotFoundError as exc:
@ -141,3 +148,11 @@ if IS_PROJECT:
LOGLEVEL = config.get('project', 'loglevel').lower() or 'info'
BASE_DIR = config.get('project', 'base_dir', fallback='.') #os.getcwd())
CONFIG = {
'project_name': PROJECT_NAME,
'production': PRODUCTION,
'secret': SECRET,
'domains': DOMAINS
}

View File

@ -66,26 +66,31 @@ def args_check(fct):
return ', '.join(array)
args_d = kwargs.get('args', {})
required = args_d.get('required', set())
args_d = kwargs.get('args', None)
if args_d is not None:
required = args_d.get('required', set())
missing = []
data = {}
missing = []
data = {}
for key in required:
data[key] = data_.pop(key, None)
if data[key] is None:
missing.append(key)
for key in required:
data[key] = data_.pop(key, None)
if data[key] is None:
missing.append(key)
if missing:
raise HTTPException(
400,
f"Missing value{plural(missing)} for: {comma_list(missing)}!")
if missing:
raise HTTPException(
400,
f"Missing value{plural(missing)} for: {comma_list(missing)}!")
optional = args_d.get('optional', set())
for key in optional:
if key in data_:
data[key] = data_[key]
optional = args_d.get('optional', set())
for key in optional:
if key in data_:
data[key] = data_[key]
else:
""" Unsafe mode, without specified arguments
"""
data = data_
kwargs['data'] = data

1
halfapi/lib/constants.py Normal file
View File

@ -0,0 +1 @@
VERBS = ('GET', 'POST', 'PUT', 'PATCH', 'DELETE')

View File

@ -3,15 +3,71 @@
lib/domain.py The domain-scoped utility functions
"""
import os
import re
import sys
import importlib
import inspect
import logging
from types import ModuleType, FunctionType
from typing import Generator, Dict, List
from typing import Any, Callable, Coroutine, Generator
from typing import Dict, List, Tuple, Iterator
import inspect
from starlette.exceptions import HTTPException
from halfapi.lib import acl
from halfapi.lib.responses import ORJSONResponse
from halfapi.lib.router import read_router
from halfapi.lib.constants import VERBS
logger = logging.getLogger("uvicorn.asgi")
VERBS = ('GET', 'POST', 'PUT', 'PATCH', 'DELETE')
class MissingAclError(Exception):
pass
class PathError(Exception):
pass
class UnknownPathParameterType(Exception):
pass
class UndefinedRoute(Exception):
pass
class UndefinedFunction(Exception):
pass
def route_decorator(fct: FunctionType, ret_type: str = 'json') -> Coroutine:
""" Returns an async function that can be mounted on a router
"""
if ret_type == 'json':
@acl.args_check
async def wrapped(request, *args, **kwargs):
fct_args_spec = inspect.getfullargspec(fct).args
fct_args = request.path_params.copy()
if 'halfapi' in fct_args_spec:
fct_args['halfapi'] = {
'user': request.user if
'user' in request else None,
'config': request.scope['config']
}
if 'data' in fct_args_spec:
fct_args['data'] = kwargs.get('data')
try:
return ORJSONResponse(fct(**fct_args))
except NotImplementedError as exc:
raise HTTPException(501) from exc
else:
raise Exception('Return type not available')
return wrapped
def get_fct_name(http_verb: str, path: str) -> str:
"""
@ -57,59 +113,53 @@ def get_fct_name(http_verb: str, path: str) -> str:
return '_'.join(fct_name)
def gen_routes(route_params: Dict, path: List, m_router: ModuleType) -> Generator:
def gen_routes(m_router: ModuleType,
verb: str,
path: List[str],
params: List[Dict]) -> Tuple[FunctionType, Dict]:
"""
Generates a tuple of the following form for a specific path:
"/path/to/route", {
"GET": {
"fct": endpoint_fct,
"params": [
{ "acl": acl_fct, [...] }
]
},
[...]
}
Returns a tuple of the function associatied to the verb and path arguments,
and the dictionary of it's acls
Parameters:
- m_router (ModuleType): The module containing the function definition
- route_params (Dict): Contains the following keys :
- one or more HTTP VERB (if none, route is not treated)
- one or zero FQTN (if none, fqtn is set to None)
- verb (str): The HTTP verb for the route (GET, POST, ...)
- path (List): The route path, as a list (each item being a level of
deepness), from the lowest level (domain) to the highest
- m_router (ModuleType): The parent router module
- params (Dict): The acl list of the following format :
[{'acl': Function, 'args': {'required': [], 'optional': []}}]
Yields:
(str, Dict): The path routes description
Returns:
(Function, Dict): The destination function and the acl dictionary
"""
d_res = {'fqtn': route_params.get('FQTN')}
if len(params) == 0:
raise MissingAclError('[{}] {}'.format(verb, '/'.join(path)))
for verb in VERBS:
params = route_params.get(verb)
if params is None:
continue
if len(params) == 0:
logger.error('No ACL for route [{%s}] %s', verb, "/".join(path))
if len(path) == 0:
logger.error('Empty path for [{%s}]', verb)
raise PathError()
try:
fct_name = get_fct_name(verb, path[-1])
fct = getattr(m_router, fct_name)
logger.debug('%s defined in %s', fct.__name__, m_router.__name__)
except AttributeError as exc:
logger.error('%s is not defined in %s', fct_name, m_router.__name__)
continue
d_res[verb] = {'fct': fct, 'params': params}
yield f"/{'/'.join([ elt for elt in path if elt ])}", d_res
fct_name = get_fct_name(verb, path[-1])
if hasattr(m_router, fct_name):
fct = getattr(m_router, fct_name)
else:
raise UndefinedFunction('{}.{}'.format(m_router.__name__, fct_name or ''))
def gen_router_routes(m_router: ModuleType, path: List[str]) -> Generator:
if not inspect.iscoroutinefunction(fct):
return route_decorator(fct), params
else:
return fct, params
def gen_router_routes(m_router: ModuleType, path: List[str]) -> \
Iterator[Tuple[str, str, Coroutine, List]]:
"""
Recursive generatore that parses a router (or a subrouter)
and yields from gen_routes
@ -121,33 +171,43 @@ def gen_router_routes(m_router: ModuleType, path: List[str]) -> Generator:
Yields:
(str, Dict): The path routes description from **gen_routes**
(str, str, Coroutine, List): A tuple containing the path, verb,
function and parameters of the route
"""
if not hasattr(m_router, 'ROUTES'):
logger.error('Missing *ROUTES* constant in *%s*', m_router.__name__)
raise Exception(f'No ROUTES constant for {m_router.__name__}')
routes = m_router.ROUTES
for subpath, route_params in routes.items():
for subpath, params in read_router(m_router).items():
path.append(subpath)
yield from gen_routes(route_params, path, m_router)
for verb in VERBS:
if verb not in params:
continue
yield ('/'.join(filter(lambda x: len(x) > 0, path)),
verb,
*gen_routes(m_router, verb, path, params[verb])
)
for subroute in params.get('SUBROUTES', []):
#logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__)
param_match = re.fullmatch('^([A-Z_]+)_([a-z]+)$', subroute)
if param_match is not None:
try:
path.append('{{{}:{}}}'.format(
param_match.groups()[0].lower(),
param_match.groups()[1]))
except AssertionError:
raise UnknownPathParameterType(subroute)
else:
path.append(subroute)
subroutes = route_params.get('SUBROUTES', [])
for subroute in subroutes:
logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__)
path.append(subroute)
try:
submod = importlib.import_module(f'.{subroute}', m_router.__name__)
yield from gen_router_routes(
importlib.import_module(f'.{subroute}', m_router.__name__),
path)
except ImportError as exc:
logger.error('Failed to import subroute **{%s}**', subroute)
raise exc
yield from gen_router_routes(submod, path)
path.pop()
path.pop()
@ -163,14 +223,14 @@ def d_domains(config) -> Dict[str, ModuleType]:
dict[str, ModuleType]
"""
if not config.has_section('domains'):
if not 'domains' in config:
return {}
try:
sys.path.append('.')
return {
domain: importlib.import_module(''.join((domain, module)))
for domain, module in config.items('domains')
for domain, module in config['domains'].items()
}
except ImportError as exc:
logger.error('Could not load a domain : %s', exc)

View File

@ -40,10 +40,17 @@ class DomainMiddleware(BaseHTTPMiddleware):
"""
domain = scope['path'].split('/')[1]
self.domains = d_domains(self.config)
self.domains = self.config.get('domains', {})
if domain in self.domains:
if len(domain) == 0:
for domain in self.domains:
self.api[domain], self.acl[domain] = api_routes(self.domains[domain])
elif domain in self.domains:
self.api[domain], self.acl[domain] = api_routes(self.domains[domain])
else:
logger.error('domain not in self.domains %s / %s',
scope['path'],
self.domains)
scope_ = scope.copy()
scope_['domains'] = self.domains
@ -57,8 +64,7 @@ class DomainMiddleware(BaseHTTPMiddleware):
current_domain = cur_path.split('/')[0]
try:
config_section = self.config.items(current_domain)
scope_['config'] = dict(config_section)
scope_['config'] = self.config.copy()
except configparser.NoSectionError:
logger.debug(
'No specific configuration for domain **%s**', current_domain)

View File

@ -140,7 +140,7 @@ class JWTAuthenticationBackend(AuthenticationBackend):
key=self.secret_key,
algorithms=[self.algorithm],
options={
'verify_signature': bool(PRODUCTION)
'verify_signature': True
})
if is_check_call:

44
halfapi/lib/router.py Normal file
View File

@ -0,0 +1,44 @@
import os
from types import ModuleType
from typing import Dict
from halfapi.lib.constants import VERBS
def read_router(m_router: ModuleType) -> Dict:
"""
Reads a module and returns a router dict
"""
if not hasattr(m_router, 'ROUTES'):
routes = {'':{}}
acls = getattr(m_router, 'ACLS') if hasattr(m_router, 'ACLS') else None
if acls is not None:
for verb in VERBS:
if not hasattr(m_router, verb.lower()):
continue
""" There is a "verb" route in the router
"""
if verb.upper() not in acls:
continue
routes[''][verb.upper()] = []
routes[''][verb.upper()] = acls[verb.upper()].copy()
routes['']['SUBROUTES'] = []
if hasattr(m_router, '__path__'):
""" Module is a package
"""
m_path = getattr(m_router, '__path__')
if isinstance(m_path, list) and len(m_path) == 1:
routes['']['SUBROUTES'] = [
elt.name
for elt in os.scandir(m_path[0])
if elt.is_dir()
]
else:
routes = getattr(m_router, 'ROUTES')
return routes

View File

@ -16,7 +16,7 @@ Exception :
from datetime import datetime
from functools import partial, wraps
import logging
from typing import Callable, List, Dict, Generator
from typing import Callable, List, Dict, Generator, Tuple
from types import ModuleType, FunctionType
from starlette.exceptions import HTTPException
@ -105,22 +105,19 @@ def gen_starlette_routes(d_domains: Dict[str, ModuleType]) -> Generator:
"""
for domain_name, m_domain in d_domains.items():
for path, d_route in gen_router_routes(m_domain, [domain_name]):
for verb in VERBS:
if verb not in d_route.keys():
continue
yield (
Route(path,
route_acl_decorator(
d_route[verb]['fct'],
d_route[verb]['params']
),
methods=[verb])
)
for path, verb, fct, params in gen_router_routes(m_domain, []):
yield (
Route(f'/{domain_name}/{path}',
route_acl_decorator(
fct,
params
),
methods=[verb])
)
def api_routes(m_dom: ModuleType) -> Generator:
def api_routes(m_dom: ModuleType) -> Tuple[Dict, Dict]:
"""
Yields the description objects for HalfAPI app routes
@ -128,7 +125,7 @@ def api_routes(m_dom: ModuleType) -> Generator:
m_dom (ModuleType): the halfapi module
Returns:
Generator(Dict)
(Dict, Dict)
"""
d_acls = {}
@ -137,6 +134,7 @@ def api_routes(m_dom: ModuleType) -> Generator:
l_params = []
for param in params:
if 'acl' not in param.keys() or not param['acl']:
continue
@ -149,13 +147,10 @@ def api_routes(m_dom: ModuleType) -> Generator:
return l_params
d_res = {}
for path, d_route in gen_router_routes(m_dom, [m_dom.__name__]):
d_res[path] = {'fqtn': d_route['fqtn'] }
for verb in VERBS:
if verb not in d_route.keys():
continue
d_res[path][verb] = str_acl(d_route[verb]['params'])
for path, verb, fct, params in gen_router_routes(m_dom, []):
if path not in d_res:
d_res[path] = {}
d_res[path][verb] = str_acl(params)
return d_res, d_acls
@ -196,7 +191,7 @@ def debug_routes():
yield Route('/halfapi/log', debug_log)
async def error_code(request: Request, *args, **kwargs):
code = request.path_params.get('code')
code = request.path_params['code']
raise HTTPException(code)
yield Route('/halfapi/error/{code:int}', error_code)

View File

@ -10,13 +10,16 @@ Constant :
SCHEMAS (starlette.schemas.SchemaGenerator)
"""
import logging
from typing import Dict
from starlette.schemas import SchemaGenerator
from starlette.exceptions import HTTPException
from .routes import gen_starlette_routes, api_acls
from .responses import ORJSONResponse
logger = logging.getLogger('uvicorn.asgi')
SCHEMAS = SchemaGenerator(
{"openapi": "3.0.0", "info": {"title": "HalfAPI", "version": "1.0"}}
)
@ -28,6 +31,20 @@ async def get_api_routes(request, *args, **kwargs):
"""
return ORJSONResponse(request.scope['api'])
def get_api_domain_routes(domain):
async def wrapped(request, *args, **kwargs):
"""
description: Returns the current API routes description (HalfAPI 0.2.1)
as a JSON object
"""
if domain in request.scope['api']:
return ORJSONResponse(request.scope['api'][domain])
else:
raise HTTPException(404)
return wrapped
async def schema_json(request, *args, **kwargs):
"""

View File

@ -11,6 +11,7 @@ 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
@ -212,8 +213,8 @@ def dummy_app():
backend=JWTAuthenticationBackend(secret_key='dummysecret')
)
return app
@pytest.fixture
@pytest.fixture
def dummy_debug_app():
app = Starlette(debug=True)
app.add_route('/',
@ -228,3 +229,75 @@ def dummy_debug_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

View File

@ -1,5 +0,0 @@
ROUTES={
'': {
'SUBROUTES': ['abc','act']
}
}

View File

@ -1,5 +0,0 @@
ROUTES={
'': {
'SUBROUTES': ['alphabet', 'pinnochio']
}
}

View File

@ -0,0 +1,19 @@
from halfapi.lib import acl
ACLS = {
'GET': [{'acl':acl.public}],
'POST': [{'acl':acl.public}],
'PATCH': [{'acl':acl.public}],
'PUT': [{'acl':acl.public}]
}
def get(test):
return str(test)
def post(test):
return str(test)
def patch(test):
return str(test)
def put(test):
return str(test)

View File

@ -1,17 +1,8 @@
from starlette.responses import PlainTextResponse
from dummy_domain import acl
ROUTES={
'': {
'GET': [{'acl':acl.public}]
},
'{test:uuid}': {
'GET': [{'acl':None}],
'POST': [{'acl':None}],
'PATCH': [{'acl':None}],
'PUT': [{'acl':None}]
}
from halfapi.lib import acl
ACLS = {
'GET': [{'acl':acl.public}]
}
async def get(request, *args, **kwargs):

View File

@ -1,2 +1,6 @@
ROUTES={
from halfapi.lib import acl
ACLS = {
'GET' : [{'acl':acl.public}]
}
def get():
raise NotImplementedError

View File

@ -2,31 +2,21 @@
import pytest
from starlette.authentication import UnauthenticatedUser
from starlette.testclient import TestClient
from halfapi.app import application
import json
def test_get_api_routes():
c = TestClient(application)
r = c.get('/')
d_r = r.json()
assert isinstance(d_r, dict)
def test_current_user(project_runner):
"""
Missing HALFAPI_SECRET to give current user route
"""
c = TestClient(application)
def test_current_user(project_runner, application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/current_user')
assert r.status_code == 200
def test_log():
c = TestClient(application)
def test_log(application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/log')
assert r.status_code == 200
def test_error():
c = TestClient(application)
def test_error(application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/error/400')
assert r.status_code == 400
r = c.get('/halfapi/error/404')
@ -34,8 +24,8 @@ def test_error():
r = c.get('/halfapi/error/500')
assert r.status_code == 500
def test_exception():
c = TestClient(application)
def test_exception(application_debug):
c = TestClient(application_debug)
try:
r = c.get('/halfapi/exception')
assert r.status_code == 500

View File

@ -0,0 +1,41 @@
import os
import sys
import importlib
import subprocess
import time
import pytest
from starlette.routing import Route
from starlette.testclient import TestClient
from halfapi.lib.domain import gen_router_routes
def test_get_route(dummy_project, application_domain, routers):
c = TestClient(application_domain)
path = verb = params = None
for path, verb, _, params in gen_router_routes(routers, []):
if len(params):
route_path = '/dummy_domain/{}'.format(path)
try:
if verb.lower() == 'get':
r = c.get(route_path)
elif verb.lower() == 'post':
r = c.post(route_path)
elif verb.lower() == 'patch':
r = c.patch(route_path)
elif verb.lower() == 'put':
r = c.put(route_path)
elif verb.lower() == 'delete':
r = c.delete(route_path)
else:
raise Exception(verb)
try:
assert r.status_code in [200, 501]
except AssertionError as exc:
print('{} [{}] {}'.format(str(r.status_code), verb, route_path))
except NotImplementedError:
pass
if not path:
raise Exception('No route generated')

View File

@ -1,18 +1,36 @@
#!/usr/bin/env python3
import importlib
from halfapi.lib.domain import VERBS, gen_router_routes
from halfapi.lib.domain import VERBS, gen_routes, gen_router_routes, MissingAclError
from types import FunctionType
def test_gen_router_routes():
from .dummy_domain import routers
for path, d_route in gen_router_routes(routers, ['dummy_domain']):
for path, verb, fct, params in gen_router_routes(routers, ['dummy_domain']):
assert isinstance(path, str)
for verb in VERBS:
if verb not in d_route.keys():
continue
route = d_route[verb]
print(f'[{verb}] {path} {route["fct"]}')
assert len(route['params']) > 0
assert hasattr(route['fct'], '__call__')
if 'fqtn' in route:
assert isinstance(route['fqtn'], str)
assert verb in VERBS
assert len(params) > 0
assert hasattr(fct, '__call__')
def test_gen_routes():
from .dummy_domain.routers.abc.alphabet import TEST_uuid
try:
gen_routes(
TEST_uuid,
'get',
['abc', 'alphabet', 'TEST_uuid', ''],
[])
except MissingAclError:
assert True
fct, params = gen_routes(
TEST_uuid,
'get',
['abc', 'alphabet', 'TEST_uuid', ''],
TEST_uuid.ACLS['GET'])
assert isinstance(fct, FunctionType)
assert isinstance(params, list)
assert len(TEST_uuid.ACLS['GET']) == len(params)

44
tests/test_lib_router.py Normal file
View File

@ -0,0 +1,44 @@
import os
from halfapi.lib.router import read_router
def test_read_router_routers():
from .dummy_domain import routers
router_d = read_router(routers)
assert '' in router_d
assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['SUBROUTES'], list)
for elt in os.scandir(routers.__path__[0]):
if elt.is_dir():
assert elt.name in router_d['']['SUBROUTES']
def test_read_router_abc():
from .dummy_domain.routers import abc
router_d = read_router(abc)
assert '' in router_d
assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['SUBROUTES'], list)
def test_read_router_alphabet():
from .dummy_domain.routers.abc import alphabet
router_d = read_router(alphabet)
assert '' in router_d
assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['SUBROUTES'], list)
def test_read_router_TEST():
from .dummy_domain.routers.abc.alphabet import TEST_uuid
router_d = read_router(TEST_uuid)
print(router_d)
assert '' in router_d
assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['GET'], list)
assert isinstance(router_d['']['POST'], list)
assert isinstance(router_d['']['PATCH'], list)
assert isinstance(router_d['']['PUT'], list)

View File

@ -5,7 +5,6 @@ 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():
@ -15,8 +14,23 @@ def test_schemas_dict_dom():
assert isinstance(schema, dict)
def test_get_api_routes(project_runner):
c = TestClient(application)
def test_get_api_routes(project_runner, application_debug):
c = TestClient(application_debug)
r = c.get('/')
d_r = r.json()
assert isinstance(d_r, dict)
def test_get_api_dummy_domain_routes(application_domain, routers):
c = TestClient(application_domain)
r = c.get('/dummy_domain')
assert r.status_code == 200
d_r = r.json()
assert isinstance(d_r, dict)
print(d_r)
assert 'abc/alphabet' in d_r
assert 'GET' in d_r['abc/alphabet']
assert len(d_r['abc/alphabet']['GET']) > 0
assert 'acl' in d_r['abc/alphabet']['GET'][0]