146 lines
5.9 KiB
Python
146 lines
5.9 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from backend.application import (
|
|
AUTHENTICATED_POST_HANDLERS,
|
|
PUBLIC_POST_HANDLERS,
|
|
RequestHandler,
|
|
)
|
|
|
|
|
|
APP_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def class_methods(path: Path, class_name: str) -> set[str]:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
owner = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.ClassDef) and node.name == class_name
|
|
)
|
|
return {
|
|
node.name
|
|
for node in owner.body
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
}
|
|
|
|
|
|
class HttpDispatchContractTests(unittest.TestCase):
|
|
@staticmethod
|
|
def handler(path: str, calls: list[str]) -> RequestHandler:
|
|
handler = RequestHandler.__new__(RequestHandler)
|
|
handler.path = path
|
|
handler.require_auth = lambda: calls.append("auth") or True
|
|
handler.require_csrf = lambda: calls.append("csrf") or True
|
|
handler.require_access = lambda method, route: (
|
|
calls.append(f"access:{method}:{route}") or True
|
|
)
|
|
return handler
|
|
|
|
def test_named_handlers_are_real_registered_post_routes(self) -> None:
|
|
all_handlers = {**PUBLIC_POST_HANDLERS, **AUTHENTICATED_POST_HANDLERS}
|
|
self.assertEqual(
|
|
set(PUBLIC_POST_HANDLERS) & set(AUTHENTICATED_POST_HANDLERS), set()
|
|
)
|
|
for path, handler_name in all_handlers.items():
|
|
with self.subTest(path=path):
|
|
route = RequestHandler.route_registry.resolve("POST", path)
|
|
self.assertIsNotNone(route)
|
|
self.assertTrue(callable(getattr(RequestHandler, handler_name)))
|
|
expected_access = "public" if path in PUBLIC_POST_HANDLERS else None
|
|
if expected_access:
|
|
self.assertEqual(route.access, expected_access)
|
|
else:
|
|
self.assertNotEqual(route.access, "public")
|
|
|
|
def test_public_post_dispatches_without_authentication(self) -> None:
|
|
for path, handler_name in PUBLIC_POST_HANDLERS.items():
|
|
with self.subTest(path=path):
|
|
calls: list[str] = []
|
|
handler = self.handler(path, calls)
|
|
handler.require_auth = lambda: (_ for _ in ()).throw(
|
|
AssertionError("public route required authentication")
|
|
)
|
|
setattr(handler, handler_name, lambda: calls.append(handler_name))
|
|
|
|
RequestHandler.do_POST(handler)
|
|
|
|
self.assertEqual(calls, [handler_name])
|
|
|
|
def test_authenticated_post_preserves_guard_order(self) -> None:
|
|
for path, handler_name in AUTHENTICATED_POST_HANDLERS.items():
|
|
with self.subTest(path=path):
|
|
calls: list[str] = []
|
|
handler = self.handler(path, calls)
|
|
setattr(handler, handler_name, lambda: calls.append(handler_name))
|
|
|
|
RequestHandler.do_POST(handler)
|
|
|
|
self.assertEqual(
|
|
calls,
|
|
["auth", "csrf", f"access:POST:{path}", handler_name],
|
|
)
|
|
|
|
def test_failed_access_never_dispatches_protected_handler(self) -> None:
|
|
path, handler_name = next(iter(AUTHENTICATED_POST_HANDLERS.items()))
|
|
calls: list[str] = []
|
|
handler = self.handler(path, calls)
|
|
handler.require_access = lambda method, route: (
|
|
calls.append(f"access:{method}:{route}") or False
|
|
)
|
|
setattr(handler, handler_name, lambda: calls.append(handler_name))
|
|
|
|
RequestHandler.do_POST(handler)
|
|
|
|
self.assertEqual(calls, ["auth", "csrf", f"access:POST:{path}"])
|
|
|
|
def test_application_composition_and_route_ownership_stay_narrow(self) -> None:
|
|
application = APP_ROOT / "backend" / "application.py"
|
|
dispatch = APP_ROOT / "backend" / "http" / "dispatch.py"
|
|
self.assertLessEqual(len(application.read_text(encoding="utf-8").splitlines()), 220)
|
|
self.assertLessEqual(len(dispatch.read_text(encoding="utf-8").splitlines()), 150)
|
|
self.assertEqual(class_methods(application, "DashboardService"), {"__init__"})
|
|
self.assertEqual(class_methods(application, "RequestHandler"), set())
|
|
self.assertEqual(
|
|
class_methods(dispatch, "ApplicationHttpDispatchMixin"),
|
|
{"_dispatch_named_handler", "do_GET", "do_POST", "do_DELETE"},
|
|
)
|
|
|
|
expected_route_owners = {
|
|
"accounts", "alerts", "auction", "dragon_tiger", "heaven", "market",
|
|
"mentor", "pools", "popularity", "review", "rotation", "screener",
|
|
"sentiment", "system", "themes",
|
|
}
|
|
route_files = {
|
|
path.parent.name: path
|
|
for path in (APP_ROOT / "backend" / "features").glob("*/routes.py")
|
|
}
|
|
self.assertEqual(set(route_files), expected_route_owners)
|
|
for path in route_files.values():
|
|
source = path.read_text(encoding="utf-8")
|
|
self.assertLessEqual(len(source.splitlines()), 120, path.name)
|
|
self.assertNotIn("backend.application", source)
|
|
self.assertNotRegex(source, r"\bSERVICE\b")
|
|
|
|
def test_application_service_methods_have_single_domain_owners(self) -> None:
|
|
owners = (
|
|
("features/system/service.py", "SystemServiceMixin", 300),
|
|
("features/accounts/application.py", "AccountApplicationMixin", 100),
|
|
("jobs/service.py", "JobServiceMixin", 80),
|
|
)
|
|
claimed: set[str] = set()
|
|
for relative, class_name, line_limit in owners:
|
|
path = APP_ROOT / "backend" / relative
|
|
methods = class_methods(path, class_name)
|
|
self.assertTrue(claimed.isdisjoint(methods))
|
|
claimed.update(methods)
|
|
self.assertLessEqual(len(path.read_text(encoding="utf-8").splitlines()), line_limit)
|
|
self.assertEqual(len(claimed), 27)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|