masq
/
halfapi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[wip] refactor half_domain

This commit is contained in:
Maxime Alves LIRMM 2021-12-07 11:42:02 +01:00
parent b4157c4a7d
commit 7001cec86e
13 changed files with 444 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
halfapi/cli/routes.py
View File

@ -14,7 +14,7 @@ from .cli import cli
from ..logging import logger from ..logging import logger
from ..lib.domain import domain_schema_dict # from ..lib.domain import domain_schema_dict
from ..lib.constants import DOMAIN_SCHEMA, ROUTE_SCHEMA from ..lib.constants import DOMAIN_SCHEMA, ROUTE_SCHEMA
from ..lib.responses import ORJSONResponse from ..lib.responses import ORJSONResponse
# from ..lib.routes import api_routes # from ..lib.routes import api_routes
@ -31,26 +31,26 @@ def routes(module, export=False, validate=False, check=False, noheader=False, sc
""" """
The "halfapi routes" command The "halfapi routes" command
""" """
try: # try:
mod = importlib.import_module(module) #  mod = importlib.import_module(module)
except ImportError as exc: # except ImportError as exc:
raise click.BadParameter('Cannot import this module', param=module) from exc #  raise click.BadParameter('Cannot import this module', param=module) from exc
if export: # if export:
click.echo(schema_to_csv(module, header=not noheader)) #  click.echo(schema_to_csv(module, header=not noheader))
if schema: # if schema:
routes_d = domain_schema_dict(mod) #  routes_d = domain_schema_dict(mod)
ROUTE_SCHEMA.validate(routes_d) #  ROUTE_SCHEMA.validate(routes_d)
click.echo(orjson.dumps(routes_d, #  click.echo(orjson.dumps(routes_d,
option=orjson.OPT_NON_STR_KEYS, #  option=orjson.OPT_NON_STR_KEYS,
default=ORJSONResponse.default_cast)) #  default=ORJSONResponse.default_cast))
if validate: # if validate:
routes_d = domain_schema_dict(mod) #  routes_d = domain_schema_dict(mod)
try: #  try:
ROUTE_SCHEMA.validate(routes_d) #  ROUTE_SCHEMA.validate(routes_d)
except Exception as exc: #  except Exception as exc:
raise exc #  raise exc

204
halfapi/half_domain.py
View File

@ -1,10 +1,25 @@
import importlib import importlib
import inspect
import os
import re
from typing import Coroutine, Dict, Iterator, List, Tuple
from types import ModuleType, FunctionType
from schema import SchemaError
from starlette.applications import Starlette from starlette.applications import Starlette
from starlette.routing import Router from starlette.routing import Router
import yaml
from .lib.constants import API_SCHEMA_DICT, ROUTER_SCHEMA, VERBS
from .half_route import HalfRoute from .half_route import HalfRoute
from .lib.routes import gen_domain_routes, gen_schema_routes from .lib import acl
from .lib.routes import JSONRoute
from .lib.domain import MissingAclError, PathError, UnknownPathParameterType, \
UndefinedRoute, UndefinedFunction, get_fct_name, route_decorator
from .lib.domain_middleware import DomainMiddleware from .lib.domain_middleware import DomainMiddleware
from .logging import logger from .logging import logger
@ -13,7 +28,8 @@ class HalfDomain(Starlette):
self.app = app self.app = app
self.m_domain = importlib.import_module(domain) self.m_domain = importlib.import_module(domain)
self.name = getattr('__name__', domain, domain) self.name = getattr(self.m_domain, '__name__', domain)
self.id = getattr(self.m_domain, '__id__')
if not router: if not router:
self.router = getattr('__router__', domain, '.routers') self.router = getattr('__router__', domain, '.routers')
@ -44,7 +60,7 @@ class HalfDomain(Starlette):
logger.info('HalfDomain creation %s %s', domain, config) logger.info('HalfDomain creation %s %s', domain, config)
super().__init__( super().__init__(
routes=gen_domain_routes(self.m_domain), routes=self.gen_domain_routes(),
middleware=[ middleware=[
(DomainMiddleware, (DomainMiddleware,
{ {
@ -80,29 +96,78 @@ class HalfDomain(Starlette):
# def schema(self): # def schema(self):
def gen_routes(self): @staticmethod
def gen_routes(m_router: ModuleType,
verb: str,
path: List[str],
params: List[Dict]) -> Tuple[FunctionType, Dict]:
""" """
Yields the Route objects for a domain Returns a tuple of the function associatied to the verb and path arguments,
and the dictionary of it's acls
Parameters: Parameters:
m_domains: ModuleType - m_router (ModuleType): The module containing the function definition
- 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
- params (Dict): The acl list of the following format :
[{'acl': Function, 'args': {'required': [], 'optional': []}}]
Returns: Returns:
Generator(HalfRoute)
"""
"""
yield HalfRoute('/',
JSONRoute(self.schema(self.m_domain)),
[{'acl': acl.public}],
'GET'
)
"""
for path, method, m_router, fct, params in self.gen_router_routes(self.m_domain, []): (Function, Dict): The destination function and the acl dictionary
yield HalfRoute(f'/{path}', fct, params, method)
"""
if len(params) == 0:
raise MissingAclError('[{}] {}'.format(verb, '/'.join(path)))
if len(path) == 0:
logger.error('Empty path for [{%s}]', verb)
raise PathError()
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(self, path: List[str]) -> \ if not inspect.iscoroutinefunction(fct):
return route_decorator(fct), params
else:
return acl.args_check(fct), params
# @staticmethod
# def gen_routes(m_router):
#  """
#  Yields the Route objects for a domain
#  Parameters:
#  m_domains: ModuleType
#  Returns:
#  Generator(HalfRoute)
#  """
#  """
#  yield HalfRoute('/',
#  JSONRoute(self.schema(self.m_domain)),
#  [{'acl': acl.public}],
#  'GET'
#  )
#  """
#  for path, method, _, fct, params \
#  in HalfDomain.gen_router_routes(m_router, []):
#  yield HalfRoute(f'/{path}', fct, params, method)
@staticmethod
def gen_router_routes(m_router, path: List[str]) -> \
Iterator[Tuple[str, str, ModuleType, Coroutine, List]]: Iterator[Tuple[str, str, ModuleType, Coroutine, List]]:
""" """
Recursive generator that parses a router (or a subrouter) Recursive generator that parses a router (or a subrouter)
@ -121,7 +186,7 @@ class HalfDomain(Starlette):
that decorates the endpoint function. that decorates the endpoint function.
""" """
for subpath, params in read_router(m_router).items(): for subpath, params in HalfDomain.read_router(m_router).items():
path.append(subpath) path.append(subpath)
for verb in VERBS: for verb in VERBS:
@ -130,7 +195,7 @@ class HalfDomain(Starlette):
yield ('/'.join(filter(lambda x: len(x) > 0, path)), yield ('/'.join(filter(lambda x: len(x) > 0, path)),
verb, verb,
m_router, m_router,
*gen_routes(m_router, verb, path, params[verb]) *HalfDomain.gen_routes(m_router, verb, path, params[verb])
) )
for subroute in params.get('SUBROUTES', []): for subroute in params.get('SUBROUTES', []):
@ -147,7 +212,7 @@ class HalfDomain(Starlette):
path.append(subroute) path.append(subroute)
try: try:
yield from gen_router_routes( yield from HalfDomain.gen_router_routes(
importlib.import_module(f'.{subroute}', m_router.__name__), importlib.import_module(f'.{subroute}', m_router.__name__),
path) path)
@ -160,6 +225,7 @@ class HalfDomain(Starlette):
path.pop() path.pop()
@staticmethod
def read_router(m_router: ModuleType) -> Dict: def read_router(m_router: ModuleType) -> Dict:
""" """
Reads a module and returns a router dict Reads a module and returns a router dict
@ -214,3 +280,99 @@ class HalfDomain(Starlette):
# TODO: Proper exception handling # TODO: Proper exception handling
logger.error(m_path) logger.error(m_path)
raise exc raise exc
def gen_domain_routes(self):
"""
Yields the Route objects for a domain
Parameters:
m_domains: ModuleType
Returns:
Generator(HalfRoute)
"""
yield HalfRoute('/',
JSONRoute(self.domain_schema()),
[{'acl': acl.public}],
'GET'
)
for path, method, m_router, fct, params in HalfDomain.gen_router_routes(self.m_domain, []):
yield HalfRoute(f'/{path}', fct, params, method)
def domain_schema_dict(self) -> Dict:
""" gen_router_routes return values as a dict
Parameters:
m_router (ModuleType): The domain routers' module
Returns:
Dict: Schema of dict is halfapi.lib.constants.DOMAIN_SCHEMA
@TODO: Should be a "router_schema_dict" function
"""
d_res = {}
for path, verb, m_router, fct, parameters in HalfDomain.gen_router_routes(self.m_router, []):
if path not in d_res:
d_res[path] = {}
if verb not in d_res[path]:
d_res[path][verb] = {}
d_res[path][verb]['callable'] = f'{m_router.__name__}:{fct.__name__}'
try:
d_res[path][verb]['docs'] = yaml.safe_load(fct.__doc__)
except AttributeError:
logger.error(
'Cannot read docstring from fct (fct=%s path=%s verb=%s', fct.__name__, path, verb)
d_res[path][verb]['acls'] = list(map(lambda elt: { **elt, 'acl': elt['acl'].__name__ },
parameters))
return d_res
def domain_schema(self) -> Dict:
schema = { **API_SCHEMA_DICT }
schema['domain'] = {
'name': self.name,
'id': self.id,
'version': getattr(self.m_domain, '__version__', ''),
'patch_release': getattr(self.m_domain, '__patch_release__', ''),
'routers': self.m_router.__name__,
'acls': tuple(getattr(self.m_acl, 'ACLS', ()))
}
schema['paths'] = self.domain_schema_dict()
return schema
# def domain_schema_list(m_router: ModuleType) -> List:
# """ Schema as list, one row by route/acl
# Parameters:
#
# m_router (ModuleType): The domain routers' module
#
# Returns:
#
# List[Tuple]: (path, verb, callable, doc, acls)
# """
# res = []
#
# for path, verb, m_router, fct, parameters in gen_router_routes(m_router, []):
# for params in parameters:
# res.append((
# path,
# verb,
# f'{m_router.__name__}:{fct.__name__}',
# params.get('acl').__name__,
# params.get('args', {}).get('required', []),
# params.get('args', {}).get('optional', []),
# params.get('out', [])
# ))
#
# return res

7
halfapi/halfapi.py
View File

@ -36,7 +36,7 @@ from .lib.jwt_middleware import JWTAuthenticationBackend
from .lib.responses import (ORJSONResponse, UnauthorizedResponse, from .lib.responses import (ORJSONResponse, UnauthorizedResponse,
NotFoundResponse, InternalServerErrorResponse, NotImplementedResponse, NotFoundResponse, InternalServerErrorResponse, NotImplementedResponse,
ServiceUnavailableResponse) ServiceUnavailableResponse)
from .lib.domain import domain_schema_dict, NoDomainsException, domain_schema from .lib.domain import NoDomainsException
from .lib.routes import gen_schema_routes, JSONRoute from .lib.routes import gen_schema_routes, JSONRoute
from .lib.schemas import schema_json, get_acls from .lib.schemas import schema_json, get_acls
from .logging import logger, config_logging from .logging import logger, config_logging
@ -105,6 +105,9 @@ class HalfAPI:
) )
for key, domain in self.config.get('domain', {}).items(): for key, domain in self.config.get('domain', {}).items():
if not isinstance(domain, dict):
continue
dom_name = domain.get('name', key) dom_name = domain.get('name', key)
if not domain.get('enabled', False): if not domain.get('enabled', False):
continue continue
@ -215,7 +218,7 @@ class HalfAPI:
res = { res = {
domain: HalfDomain.acls_route(domain) domain: HalfDomain.acls_route(domain)
for domain, domain_conf in self.config.get('domain', {}).items() for domain, domain_conf in self.config.get('domain', {}).items()
if domain_conf.get('enabled', False) if isinstance(domain_conf, dict) and domain_conf.get('enabled', False)
} }
async def wrapped(req, *args, **kwargs): async def wrapped(req, *args, **kwargs):

180
halfapi/lib/domain.py
View File

@ -18,7 +18,7 @@ from starlette.exceptions import HTTPException
from halfapi.lib import acl from halfapi.lib import acl
from halfapi.lib.responses import ORJSONResponse, ODSResponse from halfapi.lib.responses import ORJSONResponse, ODSResponse
from halfapi.lib.router import read_router # from halfapi.lib.router import read_router
from halfapi.lib.constants import VERBS from halfapi.lib.constants import VERBS
from ..logging import logger from ..logging import logger
@ -180,97 +180,97 @@ def gen_routes(m_router: ModuleType,
return acl.args_check(fct), params return acl.args_check(fct), params
def gen_router_routes(m_router: ModuleType, path: List[str]) -> \ # def gen_router_routes(m_router: ModuleType, path: List[str]) -> \
Iterator[Tuple[str, str, ModuleType, Coroutine, List]]: #  Iterator[Tuple[str, str, ModuleType, Coroutine, List]]:
""" #  """
Recursive generator that parses a router (or a subrouter) #  Recursive generator that parses a router (or a subrouter)
and yields from gen_routes #  and yields from gen_routes
# 
Parameters: #  Parameters:
# 
- m_router (ModuleType): The currently treated router module #  - m_router (ModuleType): The currently treated router module
- path (List[str]): The current path stack #  - path (List[str]): The current path stack
# 
Yields: #  Yields:
# 
(str, str, ModuleType, Coroutine, List): A tuple containing the path, verb, #  (str, str, ModuleType, Coroutine, List): A tuple containing the path, verb,
router module, function reference and parameters of the route. #  router module, function reference and parameters of the route.
Function and parameters are yielded from then gen_routes function, #  Function and parameters are yielded from then gen_routes function,
that decorates the endpoint function. #  that decorates the endpoint function.
""" #  """
# 
for subpath, params in read_router(m_router).items(): #  for subpath, params in read_router(m_router).items():
path.append(subpath) #  path.append(subpath)
# 
for verb in VERBS: #  for verb in VERBS:
if verb not in params: #  if verb not in params:
continue #  continue
yield ('/'.join(filter(lambda x: len(x) > 0, path)), #  yield ('/'.join(filter(lambda x: len(x) > 0, path)),
verb, #  verb,
m_router, #  m_router,
*gen_routes(m_router, verb, path, params[verb]) #  *gen_routes(m_router, verb, path, params[verb])
) #  )
# 
for subroute in params.get('SUBROUTES', []): #  for subroute in params.get('SUBROUTES', []):
#logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__) #  #logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__)
param_match = re.fullmatch('^([A-Z_]+)_([a-z]+)$', subroute) #  param_match = re.fullmatch('^([A-Z_]+)_([a-z]+)$', subroute)
if param_match is not None: #  if param_match is not None:
try: #  try:
path.append('{{{}:{}}}'.format( #  path.append('{{{}:{}}}'.format(
param_match.groups()[0].lower(), #  param_match.groups()[0].lower(),
param_match.groups()[1])) #  param_match.groups()[1]))
except AssertionError as exc: #  except AssertionError as exc:
raise UnknownPathParameterType(subroute) from exc #  raise UnknownPathParameterType(subroute) from exc
else: #  else:
path.append(subroute) #  path.append(subroute)
# 
try: #  try:
yield from gen_router_routes( #  yield from gen_router_routes(
importlib.import_module(f'.{subroute}', m_router.__name__), #  importlib.import_module(f'.{subroute}', m_router.__name__),
path) #  path)
# 
except ImportError as exc: #  except ImportError as exc:
logger.error('Failed to import subroute **{%s}**', subroute) #  logger.error('Failed to import subroute **{%s}**', subroute)
raise exc #  raise exc
# 
path.pop() #  path.pop()
# 
path.pop() #  path.pop()
# 
# def domain_schema_dict(m_router: ModuleType) -> Dict:
def domain_schema_dict(m_router: ModuleType) -> Dict: #  """ gen_router_routes return values as a dict
""" gen_router_routes return values as a dict #  Parameters:
Parameters: # 
#  m_router (ModuleType): The domain routers' module
m_router (ModuleType): The domain routers' module # 
#  Returns:
Returns: # 
#  Dict: Schema of dict is halfapi.lib.constants.DOMAIN_SCHEMA
Dict: Schema of dict is halfapi.lib.constants.DOMAIN_SCHEMA # 
#  @TODO: Should be a "router_schema_dict" function
@TODO: Should be a "router_schema_dict" function #  """
""" #  d_res = {}
d_res = {} # 
#  for path, verb, m_router, fct, parameters in gen_router_routes(m_router, []):
for path, verb, m_router, fct, parameters in gen_router_routes(m_router, []): #  if path not in d_res:
if path not in d_res: #  d_res[path] = {}
d_res[path] = {} # 
#  if verb not in d_res[path]:
if verb not in d_res[path]: #  d_res[path][verb] = {}
d_res[path][verb] = {} # 
#  d_res[path][verb]['callable'] = f'{m_router.__name__}:{fct.__name__}'
d_res[path][verb]['callable'] = f'{m_router.__name__}:{fct.__name__}' #  try:
try: #  d_res[path][verb]['docs'] = yaml.safe_load(fct.__doc__)
d_res[path][verb]['docs'] = yaml.safe_load(fct.__doc__) #  except AttributeError:
except AttributeError: #  logger.error(
logger.error( #  'Cannot read docstring from fct (fct=%s path=%s verb=%s', fct.__name__, path, verb)
'Cannot read docstring from fct (fct=%s path=%s verb=%s', fct.__name__, path, verb) # 
#  d_res[path][verb]['acls'] = list(map(lambda elt: { **elt, 'acl': elt['acl'].__name__ },
d_res[path][verb]['acls'] = list(map(lambda elt: { **elt, 'acl': elt['acl'].__name__ }, #  parameters))
parameters)) # 
#  return d_res
return d_res
from .constants import API_SCHEMA_DICT from .constants import API_SCHEMA_DICT
def domain_schema(m_domain: ModuleType) -> Dict: def domain_schema(m_domain: ModuleType) -> Dict:

2
halfapi/lib/routes.py
View File

@ -19,7 +19,7 @@ from types import ModuleType, FunctionType
import yaml import yaml
from .domain import gen_router_routes, domain_acls, route_decorator, domain_schema # from .domain import gen_router_routes, domain_acls, route_decorator, domain_schema
from .responses import ORJSONResponse from .responses import ORJSONResponse
from .acl import args_check from .acl import args_check
from ..half_route import HalfRoute from ..half_route import HalfRoute

2
halfapi/lib/schemas.py
View File

@ -17,7 +17,7 @@ from types import ModuleType
from starlette.schemas import SchemaGenerator from starlette.schemas import SchemaGenerator
from .. import __version__ from .. import __version__
from .domain import gen_router_routes, domain_schema_list # from .domain import gen_router_routes, domain_schema_list
from ..logging import logger from ..logging import logger
from .routes import gen_starlette_routes, api_routes, api_acls from .routes import gen_starlette_routes, api_routes, api_acls
from .responses import ORJSONResponse from .responses import ORJSONResponse

2
tests/cli/test_cli_run.py
View File

@ -10,7 +10,7 @@ def test_run_noproject(cli_runner):
print(result.stdout) print(result.stdout)
assert result.exit_code == 0 assert result.exit_code == 0
result = cli_runner.invoke(cli, ['run']) result = cli_runner.invoke(cli, ['run', '--dryrun'])
try: try:
assert result.exit_code == 0 assert result.exit_code == 0
except AssertionError as exc: except AssertionError as exc:

2
tests/conftest.py
View File

@ -278,7 +278,7 @@ def application_debug(project_runner):
'secret':'turlututu', 'secret':'turlututu',
'production':False, 'production':False,
'domain': { 'domain': {
'domain': { 'dummy_domain': {
'name': 'test_domain', 'name': 'test_domain',
'router': 'test_domain.routers', 'router': 'test_domain.routers',
'enabled': True, 'enabled': True,

1
tests/test_app.py
View File

@ -11,4 +11,5 @@ def test_halfapi_dummy_domain(dummy_domain):
config = {} config = {}
config['domain'] = {} config['domain'] = {}
config['domain'][dummy_domain['name']] = dummy_domain config['domain'][dummy_domain['name']] = dummy_domain
print(config)
halfapi = HalfAPI(config) halfapi = HalfAPI(config)

2
tests/test_dummy_project_router.py
View File

@ -8,8 +8,6 @@ from pprint import pprint
from starlette.routing import Route from starlette.routing import Route
from starlette.testclient import TestClient from starlette.testclient import TestClient
from halfapi.lib.domain import gen_router_routes
def test_get_config_route(dummy_project, application_domain): def test_get_config_route(dummy_project, application_domain):
c = TestClient(application_domain) c = TestClient(application_domain)
r = c.get('/') r = c.get('/')

102
tests/test_lib_domain.py
View File

@ -1,51 +1,51 @@
#!/usr/bin/env python3 # #!/usr/bin/env python3
import importlib # import importlib
from halfapi.lib.domain import VERBS, gen_routes, gen_router_routes, \ # from halfapi.lib.domain import VERBS, gen_routes, gen_router_routes, \
MissingAclError, domain_schema_dict, domain_schema_list #  MissingAclError, domain_schema_dict, domain_schema_list
# 
from types import FunctionType # from types import FunctionType
# 
# 
def test_gen_router_routes(): # def test_gen_router_routes():
from .dummy_domain import routers #  from .dummy_domain import routers
for path, verb, m_router, fct, params in gen_router_routes(routers, ['dummy_domain']): #  for path, verb, m_router, fct, params in gen_router_routes(routers, ['dummy_domain']):
assert isinstance(path, str) #  assert isinstance(path, str)
assert verb in VERBS #  assert verb in VERBS
assert len(params) > 0 #  assert len(params) > 0
assert hasattr(fct, '__call__') #  assert hasattr(fct, '__call__')
assert len(m_router.__file__) > 0 #  assert len(m_router.__file__) > 0
# 
# 
def test_gen_routes(): # def test_gen_routes():
from .dummy_domain.routers.abc.alphabet import TEST_uuid #  from .dummy_domain.routers.abc.alphabet import TEST_uuid
try: #  try:
gen_routes( #  gen_routes(
TEST_uuid, #  TEST_uuid,
'get', #  'get',
['abc', 'alphabet', 'TEST_uuid', ''], #  ['abc', 'alphabet', 'TEST_uuid', ''],
[]) #  [])
except MissingAclError: #  except MissingAclError:
assert True #  assert True
# 
fct, params = gen_routes( #  fct, params = gen_routes(
TEST_uuid, #  TEST_uuid,
'get', #  'get',
['abc', 'alphabet', 'TEST_uuid', ''], #  ['abc', 'alphabet', 'TEST_uuid', ''],
TEST_uuid.ACLS['GET']) #  TEST_uuid.ACLS['GET'])
# 
assert isinstance(fct, FunctionType) #  assert isinstance(fct, FunctionType)
assert isinstance(params, list) #  assert isinstance(params, list)
assert len(TEST_uuid.ACLS['GET']) == len(params) #  assert len(TEST_uuid.ACLS['GET']) == len(params)
# 
def test_domain_schema_dict(): # def test_domain_schema_dict():
from .dummy_domain import routers #  from .dummy_domain import routers
d_res = domain_schema_dict(routers) #  d_res = domain_schema_dict(routers)
# 
assert isinstance(d_res, dict) #  assert isinstance(d_res, dict)
# 
def test_domain_schema_list(): # def test_domain_schema_list():
from .dummy_domain import routers #  from .dummy_domain import routers
res = domain_schema_list(routers) #  res = domain_schema_list(routers)
# 
assert isinstance(res, list) #  assert isinstance(res, list)
assert len(res) > 0 #  assert len(res) > 0

114
tests/test_lib_router.py
View File

@ -1,57 +1,57 @@
import os # import os
import pytest # import pytest
from schema import SchemaError # from schema import SchemaError
from halfapi.lib.router import read_router # from halfapi.lib.router import read_router
from halfapi.lib.constants import ROUTER_SCHEMA, ROUTER_ACLS_SCHEMA # from halfapi.lib.constants import ROUTER_SCHEMA, ROUTER_ACLS_SCHEMA
# 
def test_read_router_routers(): # def test_read_router_routers():
from .dummy_domain import routers #  from .dummy_domain import routers
# 
router_d = read_router(routers) #  router_d = read_router(routers)
assert '' in router_d #  assert '' in router_d
assert 'SUBROUTES' in router_d[''] #  assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['SUBROUTES'], list) #  assert isinstance(router_d['']['SUBROUTES'], list)
# 
for elt in os.scandir(routers.__path__[0]): #  for elt in os.scandir(routers.__path__[0]):
if elt.is_dir(): #  if elt.is_dir():
assert elt.name in router_d['']['SUBROUTES'] #  assert elt.name in router_d['']['SUBROUTES']
# 
def test_read_router_abc(): # def test_read_router_abc():
from .dummy_domain.routers import abc #  from .dummy_domain.routers import abc
router_d = read_router(abc) #  router_d = read_router(abc)
# 
assert '' in router_d #  assert '' in router_d
assert 'SUBROUTES' in router_d[''] #  assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['SUBROUTES'], list) #  assert isinstance(router_d['']['SUBROUTES'], list)
# 
def test_read_router_alphabet(): # def test_read_router_alphabet():
from .dummy_domain.routers.abc import alphabet #  from .dummy_domain.routers.abc import alphabet
router_d = read_router(alphabet) #  router_d = read_router(alphabet)
# 
assert '' in router_d #  assert '' in router_d
assert 'SUBROUTES' in router_d[''] #  assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['SUBROUTES'], list) #  assert isinstance(router_d['']['SUBROUTES'], list)
# 
ROUTER_SCHEMA.validate(router_d) #  ROUTER_SCHEMA.validate(router_d)
# 
with pytest.raises(SchemaError): #  with pytest.raises(SchemaError):
""" Test that we cannot specify wrong method in ROUTES or ACLS #  """ Test that we cannot specify wrong method in ROUTES or ACLS
# 
TODO: Write more errors #  TODO: Write more errors
""" #  """
router_d['']['TEG'] = {} #  router_d['']['TEG'] = {}
ROUTER_SCHEMA.validate(router_d) #  ROUTER_SCHEMA.validate(router_d)
# 
def test_read_router_TEST(): # def test_read_router_TEST():
from .dummy_domain.routers.abc.alphabet import TEST_uuid #  from .dummy_domain.routers.abc.alphabet import TEST_uuid
router_d = read_router(TEST_uuid) #  router_d = read_router(TEST_uuid)
# 
print(router_d) #  print(router_d)
assert '' in router_d #  assert '' in router_d
assert 'SUBROUTES' in router_d[''] #  assert 'SUBROUTES' in router_d['']
assert isinstance(router_d['']['GET'], list) #  assert isinstance(router_d['']['GET'], list)
assert isinstance(router_d['']['POST'], list) #  assert isinstance(router_d['']['POST'], list)
assert isinstance(router_d['']['PATCH'], list) #  assert isinstance(router_d['']['PATCH'], list)
assert isinstance(router_d['']['PUT'], list) #  assert isinstance(router_d['']['PUT'], list)
# 
# 

68
tests/test_lib_routes.py
View File

@ -1,34 +1,34 @@
from starlette.routing import Route # from starlette.routing import Route
from halfapi.lib.routes import gen_starlette_routes, gen_router_routes # from halfapi.lib.routes import gen_starlette_routes, gen_router_routes
# 
def test_gen_starlette_routes(): # def test_gen_starlette_routes():
from .dummy_domain import routers #  from .dummy_domain import routers
for route in gen_starlette_routes({ #  for route in gen_starlette_routes({
'dummy_domain': routers }): #  'dummy_domain': routers }):
# 
assert isinstance(route, Route) #  assert isinstance(route, Route)
# 
import pytest # import pytest
# 
@pytest.mark.skip # @pytest.mark.skip
def test_api_routes(): # def test_api_routes():
from . import dummy_domain #  from . import dummy_domain
d_res, d_acls = api_routes(dummy_domain) #  d_res, d_acls = api_routes(dummy_domain)
assert isinstance(d_res, dict) #  assert isinstance(d_res, dict)
assert isinstance(d_acls, dict) #  assert isinstance(d_acls, dict)
# 
yielded = False #  yielded = False
# 
for path, verb, m_router, fct, params in gen_router_routes(dummy_domain, []): #  for path, verb, m_router, fct, params in gen_router_routes(dummy_domain, []):
if not yielded: #  if not yielded:
yielded = True #  yielded = True
# 
assert path in d_res #  assert path in d_res
assert verb in d_res[path] #  assert verb in d_res[path]
assert 'docs' in d_res[path][verb] #  assert 'docs' in d_res[path][verb]
assert 'acls' in d_res[path][verb] #  assert 'acls' in d_res[path][verb]
assert isinstance(d_res[path][verb]['docs'], dict) #  assert isinstance(d_res[path][verb]['docs'], dict)
assert isinstance(d_res[path][verb]['acls'], list) #  assert isinstance(d_res[path][verb]['acls'], list)
assert len(d_res[path][verb]['acls']) == len(params) #  assert len(d_res[path][verb]['acls']) == len(params)
# 
assert yielded is True #  assert yielded is True