105 lines
4.0 KiB
Python
105 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FEATURES = ROOT / "backend" / "features"
|
|
|
|
|
|
class FeatureBoundaryTests(unittest.TestCase):
|
|
def test_feature_services_do_not_import_http_or_provider_adapters(self) -> None:
|
|
forbidden = {
|
|
"server",
|
|
"tushare_client",
|
|
"ifind_client",
|
|
"chart_data_provider",
|
|
"realtime_aggregator",
|
|
}
|
|
violations = []
|
|
for path in FEATURES.rglob("*.py"):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
names = []
|
|
if isinstance(node, ast.Import):
|
|
names = [alias.name for alias in node.names]
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
names = [node.module]
|
|
for name in names:
|
|
if name.split(".")[0] in forbidden:
|
|
violations.append(f"{path.relative_to(ROOT)} -> {name}")
|
|
self.assertEqual(violations, [])
|
|
|
|
def test_backend_uses_root_compatibility_modules_only_at_declared_boundaries(self) -> None:
|
|
compatibility_modules = {
|
|
"advanced_strategies",
|
|
"alert_service",
|
|
"api_access",
|
|
"app_config",
|
|
"assistant_agent",
|
|
"chart_data_provider",
|
|
"heaven_agent",
|
|
"heaven_engine",
|
|
"ifind_client",
|
|
"llm_strategy",
|
|
"llm_stream",
|
|
"market_insights",
|
|
"mentor_agent",
|
|
"realtime_aggregator",
|
|
"screener",
|
|
"security",
|
|
"sentiment_engine",
|
|
"server",
|
|
"strategy_tracking",
|
|
"trade_journal",
|
|
"tushare_client",
|
|
}
|
|
allowed = {
|
|
"backend/application.py": {"api_access"},
|
|
"backend/features/screener/repository.py": {"sentiment_engine"},
|
|
}
|
|
violations = []
|
|
for path in (ROOT / "backend").rglob("*.py"):
|
|
relative = path.relative_to(ROOT).as_posix()
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
names = []
|
|
if isinstance(node, ast.Import):
|
|
names = [alias.name for alias in node.names]
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
names = [node.module]
|
|
for name in names:
|
|
root_name = name.split(".")[0]
|
|
if (
|
|
root_name in compatibility_modules
|
|
and root_name not in allowed.get(relative, set())
|
|
):
|
|
violations.append(f"{relative} -> {name}")
|
|
self.assertEqual(violations, [])
|
|
|
|
def test_legacy_service_modules_are_compatibility_exports_only(self) -> None:
|
|
for filename in ("alert_service.py", "trade_journal.py", "strategy_tracking.py"):
|
|
tree = ast.parse((ROOT / filename).read_text(encoding="utf-8"))
|
|
definitions = [
|
|
node for node in tree.body
|
|
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef))
|
|
]
|
|
self.assertEqual(definitions, [], filename)
|
|
|
|
def test_each_migrated_feature_owns_one_application_service(self) -> None:
|
|
expected = {
|
|
"alerts/service.py": "AlertService",
|
|
"mentor/service.py": "MentorServiceMixin",
|
|
"review/trade_journal.py": "TradeJournalService",
|
|
"screener/tracking.py": "StrategyTrackingService",
|
|
}
|
|
for relative, class_name in expected.items():
|
|
tree = ast.parse((FEATURES / relative).read_text(encoding="utf-8"))
|
|
self.assertIn(class_name, {node.name for node in tree.body if isinstance(node, ast.ClassDef)})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|