xiaobai-review/tests/test_http_dispatch.py

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), 29)


if __name__ == "__main__":
    unittest.main()