test: enforce provider construction ownership

This commit is contained in:
leefer
2026-08-02 00:42:46 +08:00
parent 053c5092bc
commit 550596f89a
3 changed files with 54 additions and 2 deletions
+19
View File
@@ -220,6 +220,25 @@
"runtime_role": "index observation fallback"
}
],
"provider_construction": [
{
"client": "TushareClient",
"owner": "backend/data/providers/tushare.py",
"compatibility_fallback": "backend/features/market/service.py"
},
{
"client": "IfindHttpClient",
"owner": "backend/data/gateway.py"
},
{
"client": "MarketChartClient",
"owner": "backend/data/gateway.py"
},
{
"client": "WebRealtimeAggregator",
"owner": "backend/data/gateway.py"
}
],
"numeric_normalization": [
{
"function": "finite_number",
+29 -2
View File
@@ -1,7 +1,9 @@
from __future__ import annotations
import ast
import unittest
from datetime import datetime, timedelta
from pathlib import Path
from backend.data import (
DataPolicyError,
@@ -45,8 +47,6 @@ class DataGatewayTests(unittest.TestCase):
self.assertIs(gateway.chart_data.ifind, gateway.ifind)
def test_server_has_no_direct_runtime_tushare_construction(self) -> None:
from pathlib import Path
source = (
Path(__file__).resolve().parents[1]
/ "backend"
@@ -57,6 +57,33 @@ class DataGatewayTests(unittest.TestCase):
self.assertEqual(source.count("TushareClient(self.token)"), 1)
self.assertIn("return gateway.tushare()", source)
def test_provider_construction_has_unique_declared_owners(self) -> None:
root = Path(__file__).resolve().parents[1]
owners = {
"EastmoneyChartClient": {"backend/data/gateway.py"},
"IfindHttpClient": {"backend/data/gateway.py"},
"IfindProvider": {"backend/data/gateway.py"},
"MarketChartClient": {"backend/data/gateway.py"},
"TushareClient": {"backend/features/market/service.py"},
"TushareProvider": {"backend/data/gateway.py"},
"WebRealtimeAggregator": {"backend/data/gateway.py"},
}
found = {name: set() for name in owners}
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):
if not isinstance(node, ast.Call):
continue
name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
if name in found:
found[name].add(relative)
self.assertEqual(found, owners)
provider_source = (root / "backend/data/providers/tushare.py").read_text(
encoding="utf-8"
)
self.assertIn("client_factory: Callable[[str], TushareClient] = TushareClient", provider_source)
def test_quality_gate_accepts_matching_daily_evidence(self) -> None:
timezone = market_timezone()
now = datetime(2026, 7, 29, 16, 0, tzinfo=timezone)
+6
View File
@@ -155,6 +155,12 @@ def build() -> dict[str, Any]:
{"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
{"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"},
],
"provider_construction": [
{"client": "TushareClient", "owner": "backend/data/providers/tushare.py", "compatibility_fallback": "backend/features/market/service.py"},
{"client": "IfindHttpClient", "owner": "backend/data/gateway.py"},
{"client": "MarketChartClient", "owner": "backend/data/gateway.py"},
{"client": "WebRealtimeAggregator", "owner": "backend/data/gateway.py"},
],
"numeric_normalization": [
{"function": "finite_number", "path": "backend/data/numbers.py"},
{"function": "non_nan_number", "path": "backend/data/numbers.py"},