118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
from pathlib import Path
|
|
|
|
import app.services.context_cache_service.mixin_cache_core as cache_core_module
|
|
import app.services.context_cache_service.mixin_profile_route_parser as route_parser_module
|
|
|
|
from app.services.context_cache_service.mixin_cache_core import ContextCacheCoreMixin
|
|
from app.services.context_cache_service.mixin_profile_basics import ContextCacheProfileBasicsMixin
|
|
from app.services.context_cache_service.mixin_profile_route_parser import ContextCacheProfileRouteParserMixin
|
|
|
|
|
|
class _FakeContextCacheProfileService(
|
|
ContextCacheProfileBasicsMixin,
|
|
ContextCacheProfileRouteParserMixin,
|
|
):
|
|
pass
|
|
|
|
|
|
class _FakeContextCacheCoreService(ContextCacheCoreMixin):
|
|
def __init__(self) -> None:
|
|
super().__init__(redis_url=None)
|
|
|
|
|
|
def _find_route(items: list[dict], route: str) -> dict:
|
|
for item in items:
|
|
if str(item.get("route") or "") == route:
|
|
return item
|
|
raise AssertionError(f"route not found in catalog: {route}")
|
|
|
|
|
|
def test_default_feature_catalog_uses_canonical_rag_ask() -> None:
|
|
items = ContextCacheProfileBasicsMixin._default_feature_catalog_items()
|
|
ai_steward_item = _find_route(items, "/ai-steward")
|
|
apis = list(ai_steward_item.get("apis") or [])
|
|
assert "/rag/ask" in apis
|
|
assert "/ai-steward/rag/ask" not in apis
|
|
|
|
|
|
def test_route_feature_summary_uses_canonical_rag_ask() -> None:
|
|
_summary, apis, _aliases = ContextCacheProfileRouteParserMixin._route_feature_summary("/ai-steward")
|
|
assert "/rag/ask" in apis
|
|
assert "/ai-steward/rag/ask" not in apis
|
|
|
|
|
|
def test_build_platform_profile_scans_backend_app_root(tmp_path, monkeypatch) -> None:
|
|
backend_root = tmp_path / "backend"
|
|
app_root = backend_root / "app"
|
|
fake_module_file = app_root / "services" / "context_cache_service" / "mixin_profile_route_parser.py"
|
|
fake_module_file.parent.mkdir(parents=True, exist_ok=True)
|
|
fake_module_file.write_text("# fake module path anchor\n", encoding="utf-8")
|
|
|
|
files = {
|
|
app_root / "api" / "demo.py": '@router.get("/demo")\nasync def demo():\n return {"ok": True}\n',
|
|
app_root / "services" / "service_a.py": "class ServiceA:\n pass\n",
|
|
app_root / "models" / "model_a.py": "class ModelA:\n pass\n",
|
|
app_root / "schemas" / "schema_a.py": "class SchemaA:\n pass\n",
|
|
app_root / "dependencies" / "dep_a.py": "def dep_a():\n return 1\n",
|
|
app_root / "core" / "core_a.py": "VALUE = 1\n",
|
|
app_root / "config" / "config_a.py": "SETTING = 'x'\n",
|
|
}
|
|
for path, content in files.items():
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
monkeypatch.setattr(route_parser_module, "__file__", str(fake_module_file))
|
|
|
|
profile = _FakeContextCacheProfileService()._build_platform_profile()
|
|
|
|
assert profile["backend"]["module_counts"] == {
|
|
"api": 1,
|
|
"services": 2,
|
|
"models": 1,
|
|
"schemas": 1,
|
|
"dependencies": 1,
|
|
"core": 1,
|
|
"config": 1,
|
|
}
|
|
assert profile["backend"]["api_route_count_estimate"] == 1
|
|
assert profile["source_scan"]["backend_root"] == str(backend_root)
|
|
assert profile["source_scan"]["app_root"] == str(app_root)
|
|
assert profile["source_scan"]["feature_catalog_source"] == "fallback"
|
|
|
|
|
|
def test_is_system_snapshot_complete_rejects_stale_source_scan(tmp_path, monkeypatch) -> None:
|
|
backend_root = tmp_path / "backend"
|
|
app_root = backend_root / "app"
|
|
fake_module_file = app_root / "services" / "context_cache_service" / "mixin_cache_core.py"
|
|
fake_module_file.parent.mkdir(parents=True, exist_ok=True)
|
|
fake_module_file.write_text("# fake module path anchor\n", encoding="utf-8")
|
|
monkeypatch.setattr(cache_core_module, "__file__", str(fake_module_file))
|
|
|
|
service = _FakeContextCacheCoreService()
|
|
|
|
stale_payload = {
|
|
"schema_version": "2026-04-01.context-v1",
|
|
"assistant_profile": {"name": "PMFlow"},
|
|
"platform_profile": {
|
|
"source_scan": {
|
|
"backend_root": str(app_root),
|
|
"app_root": str(app_root / "app"),
|
|
"workspace_root": str(tmp_path),
|
|
}
|
|
},
|
|
}
|
|
fresh_payload = {
|
|
"schema_version": cache_core_module.SYSTEM_SNAPSHOT_SCHEMA_VERSION,
|
|
"assistant_profile": {"name": "PMFlow"},
|
|
"platform_profile": {
|
|
"source_scan": {
|
|
"backend_root": str(backend_root),
|
|
"app_root": str(app_root),
|
|
"workspace_root": str(tmp_path),
|
|
}
|
|
},
|
|
}
|
|
|
|
assert service._is_system_snapshot_complete(stale_payload) is False
|
|
assert service._is_system_snapshot_complete(fresh_payload) is True
|