refactor: establish standalone application boundary

This commit is contained in:
leefer
2026-08-03 21:42:25 +08:00
parent 656f28a96d
commit d6def3af15
322 changed files with 73872 additions and 44656 deletions
+30 -8
View File
@@ -47,6 +47,27 @@ function session(role = "admin", subscribed = true) {
};
}
function waitForApplicationRuntime(page) {
return expect(page.locator("body")).toHaveAttribute("data-runtime-ready", "true");
}
function installApplicationNavigation(page) {
if (page.__xiaobaiNavigationWrapped) return;
page.__xiaobaiNavigationWrapped = true;
const goto = page.goto.bind(page);
const reload = page.reload.bind(page);
page.goto = async (...args) => {
const response = await goto(...args);
await waitForApplicationRuntime(page);
return response;
};
page.reload = async (...args) => {
const response = await reload(...args);
await waitForApplicationRuntime(page);
return response;
};
}
function mentorDirectory(role = "admin") {
const mentors = [
{
@@ -107,6 +128,7 @@ function mentorDirectory(role = "admin") {
}
async function mockApplication(page, authSession = session(), options = {}) {
installApplicationNavigation(page);
let screenerTracking = {
batches: [{
run_id: 44,
@@ -582,8 +604,8 @@ test("night mode covers the application shell and persists across reloads", asyn
const color = (selector) => getComputedStyle(document.querySelector(selector)).backgroundColor;
return {
body: color("body"),
sidebar: color(".sidebar"),
topbar: color(".topbar"),
sidebar: color(".module-nav"),
topbar: color(".app-header"),
tableHead: color("#limitTable thead th"),
};
});
@@ -1850,7 +1872,7 @@ test("mobile shell stays within the viewport", async ({ page }) => {
await expect(page.locator("#globalSearchButton")).toBeVisible();
await expect(page.locator('[data-view="limitPool"]')).toHaveClass(/mobile-active/);
const mobileShell = await page.evaluate(() => {
const header = document.querySelector(".topbar").getBoundingClientRect();
const header = document.querySelector(".app-header").getBoundingClientRect();
const main = document.querySelector(".app-main").getBoundingClientRect();
return { headerBottom: header.bottom, mainTop: main.top };
});
@@ -1949,7 +1971,7 @@ test("new review workflows render account-scoped records", async ({ page }) => {
return { labelHeight: Math.round(labelBox.height), gap: Math.round(textareaBox.top - labelBox.bottom) };
});
expect(journalSpacing).toEqual({ labelHeight: 18, gap: 6 });
await page.screenshot({ path: "test-results/review-stage17-1440.png", fullPage: true });
await page.screenshot({ path: "runtime/test-results/review-stage17-1440.png", fullPage: true });
await page.locator("#openWatchlistDialog").click();
await expect(page.locator("#watchlistDialog")).toBeVisible();
await page.locator("#watchlistSearchInput").fill("002141");
@@ -2169,7 +2191,7 @@ test("screener redesign preserves three clear workspaces across desktop and mobi
const box = await metric.boundingBox();
expect(box.height).toBeLessThanOrEqual(40);
}
await page.screenshot({ path: "test-results/screener-stage15-phase-1440.png", fullPage: true });
await page.screenshot({ path: "runtime/test-results/screener-stage15-phase-1440.png", fullPage: true });
await page.evaluate(() => {
state.screenerSetup.strategies.push({
@@ -2200,7 +2222,7 @@ test("screener redesign preserves three clear workspaces across desktop and mobi
expect(Math.abs(detailBox.y - libraryBox.y)).toBeLessThanOrEqual(1);
expect(libraryBox.width).toBeLessThan(detailBox.width);
await expect(page.locator("#curatedHealthMetrics > div")).toHaveCount(4);
await page.screenshot({ path: "test-results/screener-stage15-strategy-1440.png", fullPage: true });
await page.screenshot({ path: "runtime/test-results/screener-stage15-strategy-1440.png", fullPage: true });
await page.locator('[data-screener-mode="quant"]').click();
await expect(page.locator("#quantScoreRows .quant-score-row")).toHaveCount(5);
@@ -2219,7 +2241,7 @@ test("screener redesign preserves three clear workspaces across desktop and mobi
expect(quantRunBox.width).toBeLessThan(180);
expect((await page.locator("#quantFilterRows .quant-filter-row select").first().boundingBox()).width).toBeLessThanOrEqual(225);
await expect(page.locator('[data-screener-results-slot="quant"] > .screener-results-view')).toBeVisible();
await page.screenshot({ path: "test-results/screener-stage15-quant-1440.png", fullPage: true });
await page.screenshot({ path: "runtime/test-results/screener-stage15-quant-1440.png", fullPage: true });
await page.setViewportSize({ width: 375, height: 812 });
const overflow = await page.evaluate(() => document.documentElement.scrollWidth - window.innerWidth);
@@ -2382,7 +2404,7 @@ test("screener tracking is an internal page populated only by manual candidate a
await expect(page.locator('[data-view="screenerView"]').first()).toHaveClass(/active/);
await expect(page.locator("#trackingTableBody tr")).toHaveCount(2);
await expect(page.locator('[data-view="screenerTrackingView"]')).toHaveCount(0);
await page.screenshot({ path: "test-results/screener-stage15-tracking-1440.png", fullPage: true });
await page.screenshot({ path: "runtime/test-results/screener-stage15-tracking-1440.png", fullPage: true });
page.once("dialog", (dialog) => dialog.accept());
await page.locator('[data-remove-tracking="10"]').click();
+47
View File
@@ -0,0 +1,47 @@
from __future__ import annotations
import re
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[1]
FRONTEND_ROOT = APP_ROOT / "frontend"
PAGE_MOUNT_MARKER = " <!-- Registered page fragments mount here. -->\n"
def registered_frontend_fragments() -> list[str]:
registry = (FRONTEND_ROOT / "pages.config.js").read_text(encoding="utf-8")
return re.findall(
r'^\s*\["[^"]+", "(/pages/[^"]+/page\.html(?:\?[^"]*)?)", \[',
registry,
re.MULTILINE,
)
def registered_frontend_runtime_scripts() -> list[str]:
registry = (FRONTEND_ROOT / "pages.config.js").read_text(encoding="utf-8")
block = re.search(r"const runtimeScripts = \[(.*?)\n \];", registry, re.DOTALL)
if not block:
raise AssertionError("runtimeScripts registry is missing")
return re.findall(r'^\s*"([^"]+)",$', block.group(1), re.MULTILINE)
def assembled_frontend_document() -> str:
shell = (FRONTEND_ROOT / "index.html").read_text(encoding="utf-8")
if shell.count(PAGE_MOUNT_MARKER) != 1:
raise AssertionError("index.html must contain exactly one page mount marker")
fragments = []
for url in registered_frontend_fragments():
relative = url.split("?", 1)[0].lstrip("/")
fragments.append((FRONTEND_ROOT / relative).read_text(encoding="utf-8"))
return shell.replace(PAGE_MOUNT_MARKER, "".join(fragments))
def assembled_frontend_runtime() -> str:
chunks = []
for url in registered_frontend_runtime_scripts():
relative = url.split("?", 1)[0].lstrip("/")
if relative.startswith("vendor/"):
continue
chunks.append((FRONTEND_ROOT / relative).read_text(encoding="utf-8"))
return "\n".join(chunks)
-589
View File
@@ -1,589 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import re
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
FRONTEND_ROOT = APP_ROOT / "frontend"
ORIGINAL_STATIC = ORIGINAL_ROOT / "static"
SOURCE_RANGE = re.compile(
r"/\* PRESERVATION-SOURCE-BEGIN app\.js:(\d+)-(\d+) \*/\n"
r"(.*?)"
r"/\* PRESERVATION-SOURCE-END app\.js:\1-\2 \*/\n?",
re.DOTALL,
)
# These exact original app.js line ranges were retired in slice 11 after the
# definition-only symbols passed static, runtime, and compatibility review.
RETIRED_FRONTEND_SOURCE_RANGES = (
(3537, 3540),
(4139, 4145),
(4973, 4989),
(9022, 9027),
(9052, 9055),
)
AUDITED_FRONTEND_SOURCE_LINE_COUNT = 9283
# CR-12 through CR-14 remove only declarations repeated by a later rule under
# the same cascade context. Entries are either an exact retired fragment or an
# exact (source, replacement) pair. Optional trailing values declare the source
# count and how many leading occurrences to transform when an identical later
# copy must remain. Every other byte remains part of the accepted CSS baseline.
AUDITED_CSS_RETIREMENTS = {
"styles.css": (
".workspace-view.active-view {\n display: block;\n}\n\n",
"body.sidebar-collapsed {\n grid-template-columns: 64px minmax(0, 1fr);\n}\n\n",
".rotation-day:last-child { border-right: 0; }\n\n",
(
"#screenerView .probability-value strong,\n"
"#screenerView .probability-value small {\n"
" display: block;\n"
"}\n\n"
),
(
(
"@media (min-width: 721px) and (max-width: 1279px) {\n"
" .market-tape {\n"
" display: none;\n"
" }\n\n"
" .app-header {"
),
(
"@media (min-width: 721px) and (max-width: 1279px) {\n"
" .app-header {"
),
),
(
" .module-nav .nav-brand,\n"
" .module-nav .nav-group-label,\n"
" .module-nav .market-sub-tab,\n"
" .module-nav .sidebar-collapse-button {\n"
" display: none;\n"
" }\n\n"
),
(
" .module-nav .nav-group,\n"
" body.sidebar-collapsed .module-nav .nav-group {\n"
" display: contents;\n"
" margin: 0;\n"
" padding: 0;\n"
" border: 0;\n"
" }\n\n"
),
(
" .module-nav .module-tab.mobile-primary-tab,\n"
" body.sidebar-collapsed .module-nav .module-tab.mobile-primary-tab {\n"
" display: flex;\n"
" }\n\n"
),
" body.mentor-directory-open {\n overflow: hidden;\n }\n\n",
(
" body,\n"
" body.sidebar-collapsed {\n"
" display: block;\n"
" min-height: 100dvh;\n"
" padding-bottom: calc(68px + env(safe-area-inset-bottom));\n"
" }\n\n"
),
" .brand-block {\n height: 42px;\n }\n\n",
(
" .header-actions {\n"
" position: absolute;\n"
" inset: 56px 10px auto;\n"
" display: flex;\n"
" justify-content: space-between;\n"
" gap: 6px;\n"
" }\n\n"
),
(
" .header-date-group {\n"
" height: 42px;\n"
" min-width: 0;\n"
" flex: 1;\n"
" }\n\n"
),
(
" .header-actions > .icon-button {\n"
" width: 40px;\n"
" min-width: 40px;\n"
" min-height: 42px;\n"
" }\n\n"
),
" .overview-strip .metric:nth-of-type(1) { grid-column: 2; grid-row: 1; }\n",
" .overview-strip .metric:nth-of-type(3) { grid-column: 2; grid-row: 2; }\n",
" .curated-strategy-list {\n grid-template-columns: 1fr;\n }\n\n",
" .quant-universe-grid { grid-template-columns: 1fr 1fr; }\n",
(
" .auction-workspace-layout { grid-template-columns: 1fr; }\n",
"",
2,
1,
),
(
" .auction-unified-table-frame { min-height: 360px; max-height: none; }\n",
"",
2,
1,
),
" .theme-library-layout { display: block; min-height: 0; }\n",
(
".brand-block {\n gap: 10px;\n}\n\n",
"",
2,
1,
),
(
" html,\n"
" body {\n"
" min-width: 320px;\n"
" width: 100%;\n"
" }\n\n",
"",
2,
1,
),
(
" .module-nav .module-tab .lucide {\n"
" width: 20px;\n"
" height: 20px;\n"
" }\n\n",
"",
2,
1,
),
(
" .sentiment-trend-panel {\n"
" border-right: 0;\n"
" border-bottom: 1px solid var(--border);\n"
" }\n\n",
"",
2,
1,
),
" .auction-dataset-segments { min-width: 430px; }\n",
),
"renovation.css": (
".overview-strip .sentiment-block { padding-left: 0; }\n",
(
'.overview-strip[data-overview-expanded="true"] .metric-value '
"{ font-size: 18px; }\n\n"
),
(
".curated-strategy-list { grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 12px; }\n",
"",
2,
1,
),
(
".sentiment-detail-toolbar { margin-top: 0; }\n\n",
"",
2,
1,
),
(
".sentiment-detail-toolbar { margin-top: 0; }\n",
"",
2,
1,
),
".review-workspace .workspace-section + .workspace-section { border-top: 1px solid var(--border); }\n\n",
(
" body,\n"
" body.sidebar-collapsed {\n"
" display: block;\n"
" min-height: 100dvh;\n"
" padding-bottom: calc(68px + env(safe-area-inset-bottom));\n"
" }\n\n",
"",
2,
1,
),
(
" .app-header {\n"
" width: 100%;\n"
" height: 108px;\n"
" min-height: 108px;\n"
" position: relative;\n"
" display: flex;\n"
" align-items: flex-start;\n"
" padding: 8px 10px 0;\n"
" }\n\n",
"",
2,
1,
),
(" .brand-block { height: 42px; }\n", "", 2, 1),
(" .brand-block h1 { font-size: 16px; }\n", "", 2, 1),
(
" .header-actions { position: absolute; inset: 56px 10px auto; display: flex; justify-content: space-between; gap: 6px; }\n",
"",
2,
1,
),
(
" .header-date-group { height: 42px; min-width: 0; flex: 1; }\n",
"",
2,
1,
),
(
" .header-date-group .date-input { width: 104px; flex: 1; }\n",
"",
2,
1,
),
(
" .header-actions > .icon-button { width: 40px; min-width: 40px; min-height: 42px; }\n",
"",
2,
1,
),
(
" .module-nav,\n"
" body.sidebar-collapsed .module-nav {\n"
" width: 100%;\n"
" height: calc(64px + env(safe-area-inset-bottom));\n"
" min-height: 64px;\n"
" position: fixed;\n"
" inset: auto 0 0;\n"
" z-index: 45;\n"
" display: grid;\n"
" grid-template-columns: repeat(5, minmax(0, 1fr));\n"
" align-items: stretch;\n"
" padding: 4px 4px max(4px, env(safe-area-inset-bottom));\n"
" overflow: hidden;\n"
" border-top: 1px solid var(--border);\n"
" border-right: 0;\n"
" background: rgba(255, 255, 255, .98);\n"
" box-shadow: 0 -5px 18px rgba(16, 24, 40, .08);\n"
" }\n\n",
"",
2,
1,
),
(
" .app-main { width: 100%; min-height: calc(100dvh - 176px); margin: 0; padding: 10px 8px 20px; }\n",
"",
2,
1,
),
(
" .workspace-view,\n"
' body[data-active-view="screenerView"] .workspace-view,\n'
' body[data-active-view="mentorView"] .workspace-view,\n'
' body[data-active-view="reviewWorkspaceView"] .workspace-view { padding: 0 0 76px; }\n',
"",
2,
1,
),
(".insight-rail { gap: 12px; }\n", "", 2, 1),
("#screenerView .screener-strategy-view { gap: 12px; }\n", "", 2, 1),
(".mentor-chat-form .button { min-height: 38px; }\n\n", "", 2, 1),
(
".review-workspace .notes-history-section { grid-area: notes; }\n",
"",
3,
2,
),
(
".sentiment-stage-guide-grid article.current strong,\n"
".sentiment-stage-guide-grid article.current small { color: var(--danger); }\n",
"",
2,
1,
),
(
".review-workspace {\n"
" grid-template-columns: minmax(0, 1fr) 360px;\n"
' grid-template-areas: "watch journal" "trades journal" "notes notes";\n'
" align-items: start;\n"
" gap: 12px;\n"
"}\n",
"",
2,
1,
),
(".review-workspace .watchlist-section { grid-area: watch; }\n", "", 2, 1),
(".review-workspace .journal-section { grid-area: journal; }\n", "", 2, 1),
(".review-workspace .trade-journal-section { grid-area: trades; }\n", "", 2, 1),
),
"redesign-v2.css": (
"#dragonView .dragon-operation-table .dragon-col-reason { width: auto; }\n",
(
" .market-tape { display: none; }\n"
" .header-actions { width: 100%; }\n"
" .header-command-group { position: absolute; }\n"
),
(
" .sidebar-brand,\n"
" .module-nav .nav-group-label,\n"
" .sidebar-collapse-button,\n"
" .module-nav .market-sub-tab { display: none; }\n"
" .module-nav .nav-group,\n"
" body.sidebar-collapsed .module-nav .nav-group { display: contents; }\n"
" .module-nav .module-tab,\n"
" body.sidebar-collapsed .module-nav .module-tab { display: none; }\n"
),
" .module-nav .module-tab.mobile-primary-tab span { display: inline; }\n",
(
" .overview-strip .metric:nth-of-type(n + 4),\n"
" .overview-strip .metric-wide { display: none; }\n"
" .overview-toggle { display: none; }\n"
),
(
(
" #screenerView .quant-summary-pane .quant-universe-grid,\n"
" #screenerView .quant-formula-summary,\n"
" #screenerView .quant-execution-actions { grid-template-columns: 1fr; }\n"
" #screenerView .quant-score-row,\n"
" #screenerView .quant-filter-row { grid-template-columns: 1fr; }\n"
" #screenerTrackingView { padding: 10px; }\n"
),
(
" #screenerView .quant-summary-pane .quant-universe-grid,\n"
" #screenerView .quant-formula-summary,\n"
" #screenerView .quant-execution-actions { grid-template-columns: 1fr; }\n"
" #screenerTrackingView { padding: 10px; }\n"
),
),
),
}
# User-approved product behavior changes remain separate from code-retirement
# records. Each entry is an exact source/replacement pair. Optional trailing
# values declare the expected source count and how many leading occurrences to
# transform when one identical occurrence must remain.
AUDITED_CSS_REPLACEMENTS = {
"redesign-v2.css": (
(
(
'@media (min-width: 721px) {\n'
' body[data-active-view="dragonView"] .app-main { display: flex; flex-direction: column; overflow: hidden; }\n'
' body[data-active-view="dragonView"] .overview-strip { flex: 0 0 auto; }\n'
' body[data-active-view="dragonView"] #dragonView.active-view {\n'
' min-height: 0;\n'
' flex: 1 1 auto;\n'
' display: flex;\n'
' flex-direction: column;\n'
' }\n'
' body[data-active-view="dragonView"] .dragon-page-head-v2 { flex: 0 0 auto; }\n'
' body[data-active-view="dragonView"] .dragon-daily-content-v2,\n'
' body[data-active-view="dragonView"] .dragon-empty-state-v2 { flex: 1 1 auto; }\n'
'}\n'
),
(
'@media (min-width: 721px) {\n'
' body[data-active-view="dragonView"] .app-main { height: var(--workspace-height); min-height: 0; display: block; overflow: auto; }\n'
' body[data-active-view="dragonView"] #dragonView.active-view {\n'
' display: block;\n'
' overflow: visible;\n'
' }\n'
'}\n'
),
),
(
(
"/* The Dragon-Tiger page stays still; only the selected trader's operations scroll. */\n"
'@media (min-width: 721px) {\n'
' body[data-active-view="dragonView"] .dragon-daily-content-v2 {\n'
' min-height: 0;\n'
' display: grid;\n'
' grid-template-rows: auto auto auto minmax(150px, 1fr) auto;\n'
' overflow: hidden;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-card-stage-v2 {\n'
' flex: 0 0 auto;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-trader-detail-v2 {\n'
' min-height: 0;\n'
' display: flex;\n'
' flex-direction: column;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-detail-header {\n'
' flex: 0 0 auto;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-trader-detail .trader-operations {\n'
' min-height: 0;\n'
' flex: 1 1 auto;\n'
' overflow: auto;\n'
' overscroll-behavior: contain;\n'
' scrollbar-gutter: stable;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-unclassified-v2 {\n'
' max-height: 180px;\n'
' overflow: auto;\n'
' }\n'
),
(
'/* The Dragon-Tiger page owns vertical scrolling; wide operation tables scroll horizontally. */\n'
'@media (min-width: 721px) {\n'
' body[data-active-view="dragonView"] .dragon-daily-content-v2 {\n'
' display: block;\n'
' overflow: visible;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-trader-detail-v2 {\n'
' display: block;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-trader-detail .trader-operations {\n'
' max-height: none;\n'
' overflow: auto;\n'
' }\n\n'
' body[data-active-view="dragonView"] #dragonView .dragon-unclassified-v2 {\n'
' max-height: none;\n'
' overflow: visible;\n'
' }\n'
),
),
),
"design-system.css": (
(' [data-active-view="dragonView"],\n', "", 4, 3),
(' #dragonView.active-view,\n', ""),
(' #dragonView .dragon-page-head-v2,\n', ""),
(' #dragonView .dragon-daily-content-v2,\n', ""),
(
' #dragonView .dragon-daily-content-v2{overflow:hidden}\n'
' #dragonView .dragon-trader-detail-v2{min-height:0}\n'
' #dragonView .dragon-trader-detail .trader-operations{min-height:0;overflow:auto}\n\n',
"",
),
),
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def function_contract(path: Path, name: str) -> tuple[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
function = next(
node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name
)
body = ast.Module(body=function.body, type_ignores=[])
return (
ast.dump(function.args, include_attributes=False),
ast.dump(body, include_attributes=False),
)
def module_contract(
path: Path,
*,
excluded_definitions: set[str] | None = None,
excluded_import_modules: set[str] | None = None,
exclude_imports: bool = False,
) -> str:
excluded_definitions = excluded_definitions or set()
excluded_import_modules = excluded_import_modules or set()
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
tree.body = [
node
for node in tree.body
if not (exclude_imports and isinstance(node, (ast.Import, ast.ImportFrom)))
and not (
isinstance(node, ast.ImportFrom)
and node.module in excluded_import_modules
)
and getattr(node, "name", None) not in excluded_definitions
]
return ast.dump(tree, include_attributes=False)
def reassembled_frontend_runtime() -> str:
chunks: dict[tuple[int, int], str] = {}
for path in FRONTEND_ROOT.rglob("*.js"):
for match in SOURCE_RANGE.finditer(path.read_text(encoding="utf-8")):
source_range = (int(match.group(1)), int(match.group(2)))
if source_range in chunks:
raise AssertionError(f"duplicate app.js source range: {source_range}")
chunks[source_range] = match.group(3)
assembled: list[str] = []
next_line = 1
for (start, end), content in sorted(chunks.items()):
if start != next_line:
raise AssertionError(
f"app.js source coverage gap: expected line {next_line}, got {start}"
)
assembled.append(content)
next_line = end + 1
if next_line != AUDITED_FRONTEND_SOURCE_LINE_COUNT + 1:
raise AssertionError(
"app.js source coverage ended at "
f"{next_line - 1}, expected {AUDITED_FRONTEND_SOURCE_LINE_COUNT}"
)
return "".join(assembled)
def original_runtime_after_audited_retirements() -> str:
lines = (ORIGINAL_STATIC / "app.js").read_text(encoding="utf-8").splitlines(
keepends=True
)
retired = {
line_number
for start, end in RETIRED_FRONTEND_SOURCE_RANGES
for line_number in range(start, end + 1)
}
return "".join(
line for line_number, line in enumerate(lines, start=1) if line_number not in retired
)
def assert_frontend_runtime_matches_audited_baseline(testcase) -> None:
testcase.assertEqual(
reassembled_frontend_runtime(),
original_runtime_after_audited_retirements(),
)
def assert_moved_asset_matches(
testcase,
original_relative: str,
frontend_relative: str | None = None,
) -> None:
target_relative = frontend_relative or original_relative
if (
original_relative in AUDITED_CSS_RETIREMENTS
or original_relative in AUDITED_CSS_REPLACEMENTS
):
original = (ORIGINAL_STATIC / original_relative).read_text(encoding="utf-8")
for retired in AUDITED_CSS_RETIREMENTS.get(original_relative, ()):
source, replacement, *count_override = (
retired if isinstance(retired, tuple) else (retired, "")
)
expected_count = count_override[0] if count_override else 1
replacement_count = count_override[1] if len(count_override) > 1 else 1
testcase.assertEqual(original.count(source), expected_count, source)
original = original.replace(source, replacement, replacement_count)
for replacement in AUDITED_CSS_REPLACEMENTS.get(original_relative, ()):
source, target, *count_override = replacement
expected_count = count_override[0] if count_override else 1
replacement_count = count_override[1] if len(count_override) > 1 else expected_count
testcase.assertEqual(original.count(source), expected_count, source)
original = original.replace(source, target, replacement_count)
testcase.assertEqual(
(FRONTEND_ROOT / target_relative).read_text(encoding="utf-8"),
original,
original_relative,
)
return
testcase.assertEqual(
sha256(FRONTEND_ROOT / target_relative),
sha256(ORIGINAL_STATIC / original_relative),
original_relative,
)
def assert_page_prefix_matches(testcase, page_relative: str) -> None:
original = (ORIGINAL_STATIC / page_relative).read_text(encoding="utf-8")
migrated = (FRONTEND_ROOT / page_relative).read_text(encoding="utf-8")
testcase.assertTrue(
migrated.startswith(original.rstrip("\n") + "\n\n"),
page_relative,
)
+1 -1
View File
@@ -4,7 +4,7 @@ import tempfile
import unittest
from pathlib import Path
from alert_service import AlertService
from backend.features.alerts.service import AlertService
from database import ReviewDatabase
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from chart_data_provider import ChartDataError, EastmoneyChartClient
from backend.features.market.charts import ChartDataError, EastmoneyChartClient
from server import DashboardService
+4 -3
View File
@@ -4,6 +4,7 @@ import unittest
from pathlib import Path
from database import ReviewDatabase
from tests.frontend_test_helpers import registered_frontend_runtime_scripts
APP_ROOT = Path(__file__).resolve().parents[1]
@@ -16,9 +17,9 @@ class CleanupContractTests(unittest.TestCase):
self.assertFalse((FRONTEND_ROOT / "heaven-loading.js").exists())
def test_only_active_heaven_loading_asset_is_loaded(self) -> None:
html = (FRONTEND_ROOT / "index.html").read_text(encoding="utf-8")
self.assertIn('src="/pages/heaven/loading-v2.js', html)
self.assertNotIn('src="/heaven-loading.js', html)
scripts = registered_frontend_runtime_scripts()
self.assertTrue(any(path.startswith("/pages/heaven/loading-v2.js") for path in scripts))
self.assertFalse(any(path.startswith("/heaven-loading.js") for path in scripts))
def test_audited_definition_only_functions_stay_absent(self) -> None:
runtime = "\n".join(
+557 -16
View File
@@ -2,39 +2,581 @@ from __future__ import annotations
import re
import unittest
from collections import defaultdict
from pathlib import Path
from tests.frontend_test_helpers import assembled_frontend_document
ROOT = Path(__file__).resolve().parents[1]
STATIC = ROOT / "frontend"
TOKENS = STATIC / "shared" / "tokens.css"
LEGACY_STYLESHEETS = (
MODULE_STYLESHEETS = (
"shared/base.css",
"shared/shell.css",
"shared/auth.css",
"shared/components/controls.css",
"shared/components/navigation.css",
"shared/components/cards.css",
"shared/components/tables.css",
"shared/components/dialogs.css",
"shared/components/feedback.css",
"pages/market/foundation.css",
"pages/sentiment/foundation.css",
"pages/pools/foundation.css",
"pages/ladder/foundation.css",
"pages/rotation/foundation.css",
"pages/auction/foundation.css",
"pages/themes/foundation.css",
"pages/popularity/foundation.css",
"pages/dragon-tiger/foundation.css",
"pages/screener/foundation.css",
"pages/mentor/foundation.css",
"pages/heaven/foundation.css",
"pages/review/foundation.css",
)
RETIRED_STYLESHEETS = (
"styles/styles.css",
"styles/renovation.css",
"styles/redesign-v2.css",
"styles/design-system.css",
"styles/theme.css",
"pages/heaven/page.css",
)
PAGE_OWNERS = {
"sentimentCycleView": "pages/sentiment/foundation.css",
"limitPool": "pages/pools/foundation.css",
"brokenView": "pages/pools/foundation.css",
"downView": "pages/pools/foundation.css",
"yesterdayView": "pages/pools/foundation.css",
"performanceView": "pages/pools/foundation.css",
"ladderView": "pages/ladder/foundation.css",
"rotationView": "pages/rotation/foundation.css",
"auctionView": "pages/auction/foundation.css",
"themeLibraryView": "pages/themes/foundation.css",
"popularityView": "pages/popularity/foundation.css",
"dragonView": "pages/dragon-tiger/foundation.css",
"screenerView": "pages/screener/foundation.css",
"screenerTrackingView": "pages/screener/foundation.css",
"mentorView": "pages/mentor/foundation.css",
"heavenView": "pages/heaven/foundation.css",
"reviewWorkspaceView": "pages/review/foundation.css",
}
SHARED_ROOT_OWNERS = {
".app-header": "shared/shell.css",
".app-main": "shared/shell.css",
".module-nav": "shared/shell.css",
".overview-strip": "shared/shell.css",
".sidebar": "shared/shell.css",
".status-bar": "shared/shell.css",
".workspace-view": "shared/shell.css",
".button": "shared/components/controls.css",
".icon-button": "shared/components/controls.css",
".segment": "shared/components/navigation.css",
".data-table": "shared/components/tables.css",
".card": "shared/components/cards.css",
".dialog": "shared/components/dialogs.css",
".toast": "shared/components/feedback.css",
}
HISTORICAL_RUNTIME_CSS_BYTES = 861_341
HISTORICAL_SELECTOR_CONTEXT_RULES = 8_569
def normalize(value: str) -> str:
return re.sub(r"\s+", " ", re.sub(r"/\*.*?\*/", "", value, flags=re.DOTALL)).strip()
def normalize_context(value: str) -> str:
"""Normalize equivalent at-rule spellings before ownership checks."""
return re.sub(r"\s*([():,])\s*", r"\1", normalize(value))
def skip_space_and_comments(text: str, position: int) -> int:
while position < len(text):
if text[position].isspace():
position += 1
elif text.startswith("/*", position):
closing = text.find("*/", position + 2)
position = len(text) if closing < 0 else closing + 2
else:
break
return position
def find_delimiter(text: str, position: int) -> tuple[int, str]:
quote = ""
escaped = False
depth = 0
in_comment = False
while position < len(text):
char = text[position]
pair = text[position : position + 2]
if in_comment:
if pair == "*/":
in_comment = False
position += 2
continue
position += 1
continue
if quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == quote:
quote = ""
position += 1
continue
if pair == "/*":
in_comment = True
position += 2
continue
if char in "\"'":
quote = char
elif char in "([":
depth += 1
elif char in ")]":
depth = max(0, depth - 1)
elif depth == 0 and char in "{;":
return position, char
position += 1
return len(text), ""
def find_closing_brace(text: str, opening: int) -> int:
depth = 1
position = opening + 1
quote = ""
escaped = False
in_comment = False
while position < len(text):
char = text[position]
pair = text[position : position + 2]
if in_comment:
if pair == "*/":
in_comment = False
position += 2
continue
position += 1
continue
if quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == quote:
quote = ""
position += 1
continue
if pair == "/*":
in_comment = True
position += 2
continue
if char in "\"'":
quote = char
elif char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return position
position += 1
raise AssertionError(f"unclosed CSS block at offset {opening}")
def split_selectors(prelude: str) -> list[str]:
selectors: list[str] = []
start = 0
quote = ""
escaped = False
depth = 0
for index, char in enumerate(prelude):
if quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == quote:
quote = ""
continue
if char in "\"'":
quote = char
elif char in "([":
depth += 1
elif char in ")]":
depth = max(0, depth - 1)
elif char == "," and depth == 0:
selectors.append(normalize(prelude[start:index]))
start = index + 1
selectors.append(normalize(prelude[start:]))
return [selector for selector in selectors if selector]
def stylesheet_rule_keys(text: str, contexts: tuple[str, ...] = ()) -> list[tuple[tuple[str, ...], str]]:
keys: list[tuple[tuple[str, ...], str]] = []
position = 0
while True:
position = skip_space_and_comments(text, position)
if position >= len(text):
return keys
delimiter, kind = find_delimiter(text, position)
prelude = normalize(text[position:delimiter])
if not kind:
raise AssertionError(f"unparsed CSS tail: {prelude[:80]}")
if kind == ";":
position = delimiter + 1
continue
closing = find_closing_brace(text, delimiter)
body = text[delimiter + 1 : closing]
lowered = prelude.lower()
if lowered.startswith(("@media", "@supports", "@container", "@layer", "@starting-style", "@scope")):
keys.extend(stylesheet_rule_keys(body, contexts + (normalize_context(prelude),)))
elif lowered.startswith(("@keyframes", "@-webkit-keyframes")):
keys.append((contexts, prelude))
elif not prelude.startswith("@"):
keys.extend((contexts, selector) for selector in split_selectors(prelude))
position = closing + 1
def declaration_properties(body: str) -> frozenset[str]:
properties: set[str] = set()
start = 0
position = 0
quote = ""
escaped = False
depth = 0
in_comment = False
while position <= len(body):
char = body[position] if position < len(body) else ";"
pair = body[position : position + 2]
if in_comment:
if pair == "*/":
in_comment = False
position += 2
continue
position += 1
continue
if quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == quote:
quote = ""
position += 1
continue
if pair == "/*":
in_comment = True
position += 2
continue
if char in "\"'":
quote = char
elif char in "([":
depth += 1
elif char in ")]":
depth = max(0, depth - 1)
elif char == ";" and depth == 0:
declaration = normalize(body[start:position])
match = re.match(r"^(--[-a-zA-Z0-9_]+|[-a-zA-Z][-_a-zA-Z0-9]*)\s*:", declaration)
if match:
properties.add(match.group(1).lower())
start = position + 1
position += 1
return frozenset(properties)
def stylesheet_rules(
text: str,
contexts: tuple[str, ...] = (),
) -> list[tuple[tuple[str, ...], str, frozenset[str]]]:
rules: list[tuple[tuple[str, ...], str, frozenset[str]]] = []
position = 0
while True:
position = skip_space_and_comments(text, position)
if position >= len(text):
return rules
delimiter, kind = find_delimiter(text, position)
prelude = normalize(text[position:delimiter])
if not kind:
raise AssertionError(f"unparsed CSS tail: {prelude[:80]}")
if kind == ";":
position = delimiter + 1
continue
closing = find_closing_brace(text, delimiter)
body = text[delimiter + 1 : closing]
lowered = prelude.lower()
if lowered.startswith(("@media", "@supports", "@container", "@layer", "@starting-style", "@scope")):
rules.extend(stylesheet_rules(body, contexts + (normalize_context(prelude),)))
elif not prelude.startswith("@"):
properties = declaration_properties(body)
rules.extend((contexts, selector, properties) for selector in split_selectors(prelude))
position = closing + 1
def remove_page_scope(selector: str, view_id: str) -> str:
prefixes = (
rf":where\(#{re.escape(view_id)}\)\s+",
rf"#{re.escape(view_id)}\s+",
rf"body\[data-active-view=[\"']{re.escape(view_id)}[\"']\]\s+",
)
semantic = selector
for prefix in prefixes:
semantic = re.sub(rf"^(?P<theme>:root(?:\[[^\]]+\])?\s+)?{prefix}", r"\g<theme>", semantic)
return normalize(semantic)
def find_matching_parenthesis(value: str, opening: int) -> int:
depth = 1
quote = ""
escaped = False
for position in range(opening + 1, len(value)):
char = value[position]
if quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == quote:
quote = ""
continue
if char in "\"'":
quote = char
elif char == "(":
depth += 1
elif char == ")":
depth -= 1
if depth == 0:
return position
raise AssertionError(f"unclosed selector function: {value}")
def prune_selector_consumers(
selector: str,
source: str,
dynamic_prefixes: set[str],
) -> tuple[str, bool, bool]:
output: list[str] = []
cursor = 0
changed = False
function_pattern = re.compile(r":(is|where|has|not)\(", re.IGNORECASE)
while match := function_pattern.search(selector, cursor):
opening = match.end() - 1
closing = find_matching_parenthesis(selector, opening)
output.append(selector[cursor : match.start()])
name = match.group(1).lower()
content = selector[opening + 1 : closing]
arguments = split_selectors(content)
retained: list[str] = []
nested_changed = False
for argument in arguments:
pruned, reachable, argument_changed = prune_selector_consumers(argument, source, dynamic_prefixes)
nested_changed |= argument_changed
if reachable:
retained.append(pruned)
else:
nested_changed = True
if name != "not" and not retained:
return selector, False, True
if name == "not" and not retained:
changed = True
elif nested_changed or len(retained) != len(arguments):
output.append(f":{name}({', '.join(retained)})")
changed = True
else:
output.append(selector[match.start() : closing + 1])
cursor = closing + 1
output.append(selector[cursor:])
pruned_selector = "".join(output)
tokens = {
match.group(2)
for match in re.finditer(r"([.#])([_a-zA-Z][-_a-zA-Z0-9]*)", pruned_selector)
}
reachable = all(
token in source or any(token.startswith(prefix) for prefix in dynamic_prefixes)
for token in tokens
)
return pruned_selector, reachable, changed or not reachable
class CssGovernanceTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.html = (STATIC / "index.html").read_text(encoding="utf-8")
cls.html = assembled_frontend_document()
cls.tokens = TOKENS.read_text(encoding="utf-8")
def test_token_layer_loads_before_application_styles(self) -> None:
expected_order = (
"/shared/tokens.css",
"/styles/styles.css",
"/styles/renovation.css",
"/styles/redesign-v2.css",
"/styles/design-system.css",
"/styles/theme.css",
"/pages/heaven/page.css",
cls.modules = {
relative: (STATIC / relative).read_text(encoding="utf-8")
for relative in MODULE_STYLESHEETS
}
cls.rule_keys = {
relative: stylesheet_rule_keys(stylesheet)
for relative, stylesheet in cls.modules.items()
}
cls.rules = {
relative: stylesheet_rules(stylesheet)
for relative, stylesheet in cls.modules.items()
}
cls.runtime_source = "\n".join(
path.read_text(encoding="utf-8")
for path in STATIC.rglob("*")
if path.is_file()
and path.suffix in {".html", ".js", ".mjs"}
and "_css_core_before" not in path.parts
and path.name != "css-core-before.html"
)
positions = [self.html.index(path) for path in expected_order]
cls.dynamic_class_prefixes = {
*re.findall(r"([_a-zA-Z][-_a-zA-Z0-9]*)\$\{", cls.runtime_source),
*re.findall(r'["\']([_a-zA-Z][-_a-zA-Z0-9]*)["\']\s*\+', cls.runtime_source),
}
def test_runtime_loads_only_the_canonical_stylesheet_stack(self) -> None:
expected = ("shared/tokens.css", *MODULE_STYLESHEETS)
positions = [self.html.index(f'/{path}') for path in expected]
self.assertEqual(positions, sorted(positions))
self.assertEqual(self.html.count('<link rel="stylesheet"'), len(expected))
for path in expected:
self.assertEqual(self.html.count(f'/{path}'), 1, path)
def test_historical_layers_and_patch_file_names_are_absent(self) -> None:
for relative in RETIRED_STYLESHEETS:
self.assertFalse((STATIC / relative).exists(), relative)
self.assertNotIn(f'/{relative}', self.html)
for name in ("legacy.css", "override.css", "fix.css"):
self.assertFalse(any(STATIC.rglob(name)), name)
def test_every_module_declares_its_owner_and_is_nonempty(self) -> None:
for relative, stylesheet in self.modules.items():
owner = Path(relative).parent.name if "/pages/" in f"/{relative}" else Path(relative).stem
if relative.startswith("shared/components/"):
owner = Path(relative).stem
self.assertTrue(
stylesheet.startswith(
f"/* Canonical CSS owner: {owner}. Historical layers consolidated 2026-08-02. */"
),
relative,
)
self.assertIn("{", stylesheet, relative)
def test_modules_do_not_contain_empty_declarations(self) -> None:
for relative, stylesheet in self.modules.items():
self.assertNotRegex(stylesheet, r"(?m)^\s*[-_a-zA-Z0-9]+:\s*;\s*$", relative)
def test_selector_and_context_have_exactly_one_owner(self) -> None:
owners: dict[tuple[tuple[str, ...], str], list[str]] = defaultdict(list)
for relative, keys in self.rule_keys.items():
for key in keys:
owners[key].append(relative)
duplicates = {
f"{' > '.join(contexts) or '<root>'} :: {selector}": paths
for (contexts, selector), paths in owners.items()
if len(paths) != 1
}
self.assertEqual(duplicates, {})
def test_page_ids_only_appear_in_their_owner_stylesheet(self) -> None:
for relative, stylesheet in self.modules.items():
for view_id, owner in PAGE_OWNERS.items():
if re.search(rf"(?:#{re.escape(view_id)}|data-active-view=[\"']{re.escape(view_id)})", stylesheet):
self.assertIn(relative, (owner, "shared/shell.css"), f"{view_id} leaked into {relative}")
def test_unscoped_shared_roots_stay_in_their_shared_owner(self) -> None:
for relative, keys in self.rule_keys.items():
for contexts, selector in keys:
owner = SHARED_ROOT_OWNERS.get(selector)
page_scoped = any(any(f"#{view_id}" in context for view_id in PAGE_OWNERS) for context in contexts)
if owner and not page_scoped:
self.assertEqual(relative, owner, f"{selector} leaked into {relative}")
def test_historical_shell_aliases_cannot_return(self) -> None:
class_tokens = {
token
for value in re.findall(r'class=["\']([^"\']*)["\']', self.html)
for token in value.split()
}
self.assertTrue({"app-header", "module-nav", "overview-strip", "status-bar"} <= class_tokens)
self.assertTrue({"topbar", "statusbar", "mktstrip", "brand", "logo"}.isdisjoint(class_tokens))
selector_text = "\n".join(
selector
for keys in self.rule_keys.values()
for _contexts, selector in keys
)
for alias in ("topbar", "statusbar", "mktstrip"):
self.assertNotRegex(selector_text, rf"(?<![-_a-zA-Z0-9])\.{alias}(?![-_a-zA-Z0-9])")
def test_special_shared_and_page_rules_have_one_owner(self) -> None:
exact_owners = {
".lucide": "shared/components/controls.css",
".visually-hidden": "shared/base.css",
}
for selector, expected_owner in exact_owners.items():
owners = [
relative
for relative, keys in self.rule_keys.items()
if any(candidate == selector for _contexts, candidate in keys)
]
self.assertEqual(owners, [expected_owner], selector)
for relative, keys in self.rule_keys.items():
for _contexts, selector in keys:
if ".wentian-v2-dialog" in selector:
self.assertEqual(relative, "pages/heaven/foundation.css", selector)
if re.search(r"\.admin-dialog\s+\.model-", selector):
self.assertEqual(relative, "shared/components/dialogs.css", selector)
def test_page_scope_variants_never_override_the_same_property(self) -> None:
conflicts: dict[str, list[str]] = defaultdict(list)
for view_id, owner in PAGE_OWNERS.items():
variants: dict[tuple[tuple[str, ...], str], list[tuple[str, frozenset[str]]]] = defaultdict(list)
for contexts, selector, properties in self.rules[owner]:
semantic = remove_page_scope(selector, view_id)
if semantic != selector:
variants[(contexts, semantic)].append((selector, properties))
for (contexts, semantic), rows in variants.items():
unique_selectors = {selector for selector, _properties in rows}
if len(unique_selectors) < 2:
continue
for index, (left_selector, left_properties) in enumerate(rows):
for right_selector, right_properties in rows[index + 1 :]:
if left_selector == right_selector:
continue
overlap = sorted(left_properties & right_properties)
if overlap:
context = " > ".join(contexts) or "<root>"
conflicts[owner].append(
f"{context} :: {semantic} :: {left_selector} <> {right_selector}: {overlap}"
)
details = "\n".join(
f"{owner}: {detail}"
for owner, owner_conflicts in conflicts.items()
for detail in owner_conflicts
)
self.assertFalse(conflicts, details)
def test_every_selector_has_a_runtime_consumer(self) -> None:
stale: list[str] = []
for relative, keys in self.rule_keys.items():
for contexts, selector in keys:
if selector.lower().startswith(("@keyframes", "@-webkit-keyframes")):
continue
_pruned, reachable, changed = prune_selector_consumers(
selector,
self.runtime_source,
self.dynamic_class_prefixes,
)
if not reachable or changed:
context = " > ".join(contexts) or "<root>"
stale.append(f"{relative}: {context} :: {selector}")
self.assertEqual(stale, [])
def test_runtime_css_is_smaller_and_has_fewer_rules(self) -> None:
total_bytes = sum((STATIC / path).stat().st_size for path in MODULE_STYLESHEETS)
total_rules = sum(len(keys) for keys in self.rule_keys.values())
self.assertLess(total_bytes, HISTORICAL_RUNTIME_CSS_BYTES)
self.assertLess(total_rules, HISTORICAL_SELECTOR_CONTEXT_RULES)
def test_token_file_has_three_layer_contract(self) -> None:
for heading in (
@@ -57,13 +599,12 @@ class CssGovernanceTests(unittest.TestCase):
def test_light_and_dark_semantics_share_one_owner(self) -> None:
self.assertIn(':root[data-theme="dark"] {', self.tokens)
global_root = re.compile(r'(?m)^:root(?:\[data-theme="dark"\])?\s*\{')
for filename in LEGACY_STYLESHEETS:
stylesheet = (STATIC / filename).read_text(encoding="utf-8")
for filename, stylesheet in self.modules.items():
self.assertIsNone(global_root.search(stylesheet), filename)
def test_wentian_tokens_remain_isolated(self) -> None:
self.assertNotRegex(self.tokens, r"--wt-[a-z0-9-]+\s*:")
wentian = (STATIC / "pages" / "heaven" / "page.css").read_text(encoding="utf-8")
wentian = self.modules["pages/heaven/foundation.css"]
self.assertRegex(wentian, r"--wt-[a-z0-9-]+\s*:")
def test_compatibility_aliases_cover_historical_layers(self) -> None:
+1 -1
View File
@@ -5,7 +5,7 @@ from datetime import datetime, timedelta
from pathlib import Path
from database import ReviewDatabase
from screener import (
from backend.features.screener.engine import (
ADVANCED_CURATED_STRATEGIES,
CURATED_STRATEGIES,
FACTOR_FIELDS,
+28 -40
View File
@@ -7,6 +7,27 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
FEATURES = ROOT / "backend" / "features"
RETIRED_ROOT_MODULES = {
"advanced_strategies",
"alert_service",
"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",
"strategy_tracking",
"trade_journal",
"tushare_client",
}
class FeatureBoundaryTests(unittest.TestCase):
@@ -32,34 +53,7 @@ class FeatureBoundaryTests(unittest.TestCase):
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"},
}
def test_backend_does_not_import_retired_root_modules(self) -> None:
violations = []
for path in (ROOT / "backend").rglob("*.py"):
relative = path.relative_to(ROOT).as_posix()
@@ -72,21 +66,15 @@ class FeatureBoundaryTests(unittest.TestCase):
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())
):
if root_name in RETIRED_ROOT_MODULES:
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_retired_root_modules_are_absent(self) -> None:
present = sorted(
name for name in RETIRED_ROOT_MODULES if (ROOT / f"{name}.py").exists()
)
self.assertEqual(present, [])
def test_each_migrated_feature_owns_one_application_service(self) -> None:
expected = {
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from heaven_engine import build_five_phase_field
from backend.features.heaven.engine import build_five_phase_field
class FivePhaseFrameworkTests(unittest.TestCase):
+227 -36
View File
@@ -4,9 +4,14 @@ import json
import re
import unittest
from pathlib import Path
from unittest.mock import patch
from tests.preservation_helpers import reassembled_frontend_runtime
from tests.frontend_test_helpers import (
PAGE_MOUNT_MARKER,
assembled_frontend_document,
assembled_frontend_runtime,
registered_frontend_fragments,
registered_frontend_runtime_scripts,
)
ROOT = Path(__file__).resolve().parents[1]
@@ -16,45 +21,64 @@ STATIC = ROOT / "frontend"
class FrontendBoundaryTests(unittest.TestCase):
def test_shared_api_is_the_only_application_fetch_exit(self) -> None:
fetch_files = []
for path in STATIC.rglob("*.js"):
for path in STATIC.rglob("*"):
if "vendor" in path.parts:
continue
if path.suffix not in {".js", ".mjs"}:
continue
if re.search(r"\bfetch\s*\(", path.read_text(encoding="utf-8")):
fetch_files.append(path.relative_to(STATIC).as_posix())
self.assertEqual(fetch_files, ["shared/api.js"])
self.assertEqual(fetch_files, ["bootstrap.js", "shared/api.js"])
bootstrap = (STATIC / "bootstrap.js").read_text(encoding="utf-8")
self.assertIn("fetch(fragment.url", bootstrap)
self.assertNotIn('"/api/', bootstrap)
def test_shared_dependencies_load_before_application(self) -> None:
html = (STATIC / "index.html").read_text(encoding="utf-8")
ui_position = html.index('/shared/ui-core.js')
components_position = html.index('/shared/components.js')
pages_position = html.index('/pages.config.js')
runtime_position = html.index('/pages/runtime.js')
state_position = html.index('/shared/state.js')
api_position = html.index('/shared/api.js')
shell_position = html.index('/shared/shell.js')
app_position = html.index('/app.js')
scripts = registered_frontend_runtime_scripts()
ui_position = scripts.index('/shared/ui-core.js')
components_position = next(index for index, value in enumerate(scripts) if value.startswith('/shared/components.js'))
runtime_position = next(index for index, value in enumerate(scripts) if value.startswith('/pages/runtime.js'))
state_position = next(index for index, value in enumerate(scripts) if value.startswith('/shared/state.js'))
api_position = next(index for index, value in enumerate(scripts) if value.startswith('/shared/api.js'))
shell_position = next(index for index, value in enumerate(scripts) if value.startswith('/shared/shell.js'))
app_position = next(index for index, value in enumerate(scripts) if value.startswith('/app.js'))
self.assertLess(ui_position, components_position)
self.assertLess(components_position, pages_position)
self.assertLess(pages_position, state_position)
self.assertLess(pages_position, runtime_position)
self.assertLess(components_position, runtime_position)
self.assertLess(runtime_position, state_position)
self.assertLess(state_position, api_position)
self.assertLess(api_position, shell_position)
self.assertLess(shell_position, app_position)
def test_application_state_is_created_through_shared_boundary(self) -> None:
app = reassembled_frontend_runtime()
app = assembled_frontend_runtime()
self.assertIn("const state = window.XiaobaiState.create({", app)
self.assertNotIn("const state = {", app)
def test_candidate_runtime_reassembly_does_not_require_original_static(self) -> None:
with patch(
"tests.preservation_helpers.ORIGINAL_STATIC",
ROOT / "missing-original-static",
):
app = reassembled_frontend_runtime()
def test_current_runtime_assembly_uses_registered_first_party_sources(self) -> None:
expected = "\n".join(
(STATIC / url.split("?", 1)[0].lstrip("/")).read_text(encoding="utf-8")
for url in registered_frontend_runtime_scripts()
if not url.startswith("/vendor/")
)
app = assembled_frontend_runtime()
self.assertEqual(app, expected)
self.assertIn("function openView(", app)
def test_historical_frontend_replay_scaffolding_is_retired(self) -> None:
runtime = "\n".join(
path.read_text(encoding="utf-8")
for path in STATIC.rglob("*.js")
if "vendor" not in path.parts
)
helper = (ROOT / "tests" / "frontend_test_helpers.py").read_text(
encoding="utf-8"
)
self.assertNotIn("PRESERVATION-SOURCE-", runtime)
self.assertNotIn("AUDITED_CSS_RETIREMENTS", helper)
self.assertNotIn("AUDITED_CSS_REPLACEMENTS", helper)
self.assertNotIn("AUDITED_FRONTEND_SOURCE_LINE_COUNT", helper)
self.assertFalse((ROOT / "tools" / "split_frontend_runtime.py").exists())
def test_runtime_page_registry_matches_governance_registry(self) -> None:
expected = json.loads(
(ROOT / "config" / "pages.config.json").read_text(encoding="utf-8")
@@ -81,19 +105,20 @@ class FrontendBoundaryTests(unittest.TestCase):
self.assertEqual(actual, expected)
def test_shell_owns_navigation_and_page_mounting(self) -> None:
app = reassembled_frontend_runtime()
app = assembled_frontend_runtime()
entry = (STATIC / "app.js").read_text(encoding="utf-8")
shell = (STATIC / "shared" / "shell.js").read_text(encoding="utf-8")
self.assertNotIn("function syncNavigationState", app)
self.assertNotIn("function initializeApplicationShell", app)
self.assertIn("function syncNavigation(viewId)", shell)
self.assertIn("function mount(viewId, mountOptions = {})", shell)
self.assertIn("function openModalDialog(dialog)", shell)
self.assertNotIn('document.querySelectorAll(".module-tab").forEach', app)
self.assertNotIn('document.querySelectorAll(".module-tab").forEach', entry)
def test_every_registered_view_has_one_feature_page_module(self) -> None:
html = (STATIC / "index.html").read_text(encoding="utf-8")
runtime_position = html.index('/pages/runtime.js')
app_position = html.index('/app.js')
scripts = registered_frontend_runtime_scripts()
runtime_position = next(index for index, value in enumerate(scripts) if value.startswith('/pages/runtime.js'))
app_position = next(index for index, value in enumerate(scripts) if value.startswith('/app.js'))
expected = {
page["id"]: page["feature"]
for page in json.loads(
@@ -104,9 +129,11 @@ class FrontendBoundaryTests(unittest.TestCase):
actual: dict[str, str] = {}
for path in (STATIC / "pages").glob("*/page.js"):
script_url = f'/pages/{path.parent.name}/page.js'
self.assertIn(script_url, html)
self.assertLess(runtime_position, html.index(script_url))
self.assertLess(html.index(script_url), app_position)
script_position = next(
index for index, value in enumerate(scripts) if value.startswith(script_url)
)
self.assertLess(runtime_position, script_position)
self.assertLess(script_position, app_position)
script = path.read_text(encoding="utf-8")
for match in re.finditer(
r'XiaobaiPageModules\.register\("([^"]+)",\s*\[(.*?)\]',
@@ -119,21 +146,185 @@ class FrontendBoundaryTests(unittest.TestCase):
actual[view_id] = feature
self.assertEqual(actual, expected)
def test_index_is_shell_only_and_has_one_bootstrap_entry(self) -> None:
html = (STATIC / "index.html").read_text(encoding="utf-8")
self.assertEqual(html.count(PAGE_MOUNT_MARKER), 1)
self.assertNotRegex(
html,
r'<section id="(?:[A-Za-z][A-Za-z0-9_-]*View|limitPool)" class="workspace-view',
)
self.assertEqual(
re.findall(r'<script[^>]+src="([^"]+)"', html),
["/bootstrap.js?v=20260803-2"],
)
def test_registered_fragments_are_the_only_page_markup_owners(self) -> None:
registry = (STATIC / "pages.config.js").read_text(encoding="utf-8")
rows = re.findall(
r'^\s*\["([^"]+)", "(/pages/[^"]+/page\.html\?[^\"]+)", \[([^\]]+)\]\],$',
registry,
re.MULTILINE,
)
self.assertEqual([row[1] for row in rows], registered_frontend_fragments())
self.assertEqual(len(rows), 12)
actual_views = []
for feature, url, declared_source in rows:
relative = url.split("?", 1)[0].lstrip("/")
fragment = (STATIC / relative).read_text(encoding="utf-8")
self.assertNotIn("<script", fragment, relative)
declared = re.findall(r'"([A-Za-z][A-Za-z0-9_-]+)"', declared_source)
roots = re.findall(
r'<section id="([A-Za-z][A-Za-z0-9_-]*View|limitPool)" class="workspace-view',
fragment,
)
self.assertEqual(roots, declared, feature)
actual_views.extend(roots)
expected = [
page["id"]
for page in json.loads(
(ROOT / "config" / "pages.config.json").read_text(encoding="utf-8")
)["pages"]
]
expected.append("screenerTrackingView")
self.assertEqual(set(actual_views), set(expected))
self.assertEqual(len(actual_views), len(set(actual_views)))
def test_runtime_script_order_has_one_registered_owner(self) -> None:
scripts = registered_frontend_runtime_scripts()
self.assertEqual(len(scripts), len(set(scripts)))
self.assertEqual(scripts[-1].split("?", 1)[0], "/app.js")
for url in scripts:
self.assertTrue((STATIC / url.split("?", 1)[0].lstrip("/")).is_file(), url)
bootstrap = (STATIC / "bootstrap.js").read_text(encoding="utf-8")
self.assertIn("registry.fragments.map(loadPageFragment)", bootstrap)
self.assertIn("loadRuntimeScripts(registry.runtimeScripts)", bootstrap)
self.assertIn("script.async = false", bootstrap)
self.assertIn('document.body.dataset.runtimeReady = "true"', bootstrap)
self.assertEqual(
assembled_frontend_document().count('class="workspace-view'),
17,
)
def test_page_lifecycle_is_owned_outside_application_monolith(self) -> None:
app = reassembled_frontend_runtime()
entry = (STATIC / "app.js").read_text(encoding="utf-8")
runtime = (STATIC / "pages" / "runtime.js").read_text(encoding="utf-8")
start = app.index("function openView(")
end = app.index("\nfunction initializeAutoTableSorting", start)
open_view = app[start:end]
start = entry.index("function openView(")
end = entry.index("\n\nif (document.readyState", start)
open_view = entry[start:end]
self.assertIn("pageModules.beforeMount(viewId, previousView);", open_view)
self.assertIn("pageModules.afterMount(viewId, previousView);", open_view)
self.assertNotRegex(open_view, r'viewId\s*[!=]==?\s*"')
self.assertIn("function beforeMount(viewId, previousView)", runtime)
self.assertIn("function afterMount(viewId, previousView)", runtime)
def test_application_entry_only_coordinates_startup(self) -> None:
entry = (STATIC / "app.js").read_text(encoding="utf-8")
self.assertLessEqual(len(entry.splitlines()), 120)
for owned_symbol in (
"XiaobaiState.create",
"XiaobaiAPI.configure",
"function loadDashboard(",
"function openAdminSettings(",
"function syncThemeControl(",
"function initializeAutoTableSorting(",
):
with self.subTest(owned_symbol=owned_symbol):
self.assertNotIn(owned_symbol, entry)
binding = re.search(r"function bindEvents\(\) \{(.*?)\n\}", entry, re.DOTALL)
self.assertIsNotNone(binding)
self.assertIn("pageModules.bind();", binding.group(1))
self.assertNotIn("addEventListener", binding.group(1))
self.assertNotIn("querySelector", binding.group(1))
def test_shared_runtime_responsibilities_have_single_owners(self) -> None:
owners = {
"context.js": "XiaobaiState.create",
"application.js": "XiaobaiAPI.configure",
"dashboard.js": "function loadDashboard(",
"admin.js": "function openAdminSettings(",
"theme.js": "function syncThemeControl(",
"table.js": "function initializeAutoTableSorting(",
}
runtime_sources = {
path.name: path.read_text(encoding="utf-8")
for path in (STATIC / "shared").glob("*.js")
}
for owner, symbol in owners.items():
with self.subTest(owner=owner, symbol=symbol):
self.assertIn(symbol, runtime_sources[owner])
self.assertEqual(
[name for name, source in runtime_sources.items() if symbol in source],
[owner],
)
def test_each_feature_owns_its_control_binding(self) -> None:
expected = {
"auction": "bindAuctionEvents",
"dragon-tiger": "bindDragonTigerEvents",
"heaven": "bindHeavenEvents",
"ladder": "bindLadderEvents",
"mentor": "bindMentorEvents",
"pools": "bindPoolEvents",
"popularity": "bindPopularityEvents",
"review": "bindReviewEvents",
"rotation": "bindRotationEvents",
"screener": "bindScreenerEvents",
"sentiment": "bindSentimentEvents",
"themes": "bindThemeLibraryEvents",
}
for feature, binder in expected.items():
with self.subTest(feature=feature):
source = (STATIC / "pages" / feature / "page.js").read_text(encoding="utf-8")
self.assertIn(f"bind: {binder}", source)
self.assertIn(f"function {binder}()", source)
market = (STATIC / "pages" / "market" / "bindings.js").read_text(encoding="utf-8")
self.assertIn("function bindMarketEvents()", market)
def test_market_runtime_has_narrow_registered_owners(self) -> None:
expected = [
"/pages/market/breadth.js",
"/pages/market/charts.js",
"/pages/market/entity-detail.js",
"/pages/market/stock-detail.js",
"/pages/market/preview.js",
"/pages/market/search.js",
"/pages/market/bindings.js",
]
registered = [
url.split("?", 1)[0]
for url in registered_frontend_runtime_scripts()
if url.startswith("/pages/market/")
]
self.assertEqual(registered, expected)
self.assertFalse((STATIC / "pages" / "market" / "runtime.js").exists())
owners = {
"breadth.js": "function renderMarketBreadth(",
"charts.js": "function currentChartPalette(",
"entity-detail.js": "function openEntityDetail(",
"stock-detail.js": "function openStock(",
"preview.js": "function showStockPreview(",
"search.js": "function openGlobalSearch(",
"bindings.js": "function bindMarketEvents(",
}
sources = {
path.name: path.read_text(encoding="utf-8")
for path in (STATIC / "pages" / "market").glob("*.js")
}
for owner, symbol in owners.items():
self.assertEqual(
[name for name, source in sources.items() if symbol in source],
[owner],
)
for relative in expected:
source = (STATIC / relative.lstrip("/")).read_text(encoding="utf-8")
self.assertLessEqual(len(source.splitlines()), 500, relative)
def test_shared_empty_state_component_is_used_by_multiple_features(self) -> None:
components = (STATIC / "shared" / "components.js").read_text(encoding="utf-8")
app = reassembled_frontend_runtime()
app = assembled_frontend_runtime()
self.assertIn("function emptyStateHtml(message, options = {})", components)
self.assertIn("function renderEmptyState(target, message, options = {})", components)
self.assertGreaterEqual(app.count("renderEmptyState("), 8)
+57 -47
View File
@@ -5,7 +5,10 @@ import unittest
from html.parser import HTMLParser
from pathlib import Path
from tests.preservation_helpers import reassembled_frontend_runtime
from tests.frontend_test_helpers import (
assembled_frontend_runtime,
assembled_frontend_document,
)
STATIC_DIR = Path(__file__).resolve().parents[1] / "frontend"
@@ -23,13 +26,21 @@ class IdCollector(HTMLParser):
class FrontendContractTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.html = (STATIC_DIR / "index.html").read_text(encoding="utf-8")
cls.script = reassembled_frontend_runtime()
cls.html = assembled_frontend_document()
cls.registry = (STATIC_DIR / "pages.config.js").read_text(encoding="utf-8")
cls.script = assembled_frontend_runtime()
cls.shell = (STATIC_DIR / "shared" / "shell.js").read_text(encoding="utf-8")
cls.ui_core = (STATIC_DIR / "shared" / "ui-core.js").read_text(encoding="utf-8")
cls.design_system = (STATIC_DIR / "styles" / "design-system.css").read_text(encoding="utf-8")
cls.theme = (STATIC_DIR / "styles" / "theme.css").read_text(encoding="utf-8")
cls.tokens = (STATIC_DIR / "shared" / "tokens.css").read_text(encoding="utf-8")
cls.auth_styles = (STATIC_DIR / "shared" / "auth.css").read_text(encoding="utf-8")
cls.feedback_styles = (STATIC_DIR / "shared" / "components" / "feedback.css").read_text(encoding="utf-8")
cls.sentiment_styles = (STATIC_DIR / "pages" / "sentiment" / "foundation.css").read_text(encoding="utf-8")
cls.screener_styles = (STATIC_DIR / "pages" / "screener" / "foundation.css").read_text(encoding="utf-8")
cls.mentor_styles = (STATIC_DIR / "pages" / "mentor" / "foundation.css").read_text(encoding="utf-8")
cls.heaven_styles = (STATIC_DIR / "pages" / "heaven" / "foundation.css").read_text(encoding="utf-8")
cls.review_styles = (STATIC_DIR / "pages" / "review" / "foundation.css").read_text(encoding="utf-8")
cls.theme_library_styles = (STATIC_DIR / "pages" / "themes" / "foundation.css").read_text(encoding="utf-8")
cls.base_styles = (STATIC_DIR / "shared" / "base.css").read_text(encoding="utf-8")
collector = IdCollector()
collector.feed(cls.html)
cls.ids = collector.ids
@@ -62,20 +73,21 @@ class FrontendContractTests(unittest.TestCase):
self.assertIn(f'data-view="{view_id}"', self.html)
for endpoint in ("/api/auction?", "/api/themes?", "/api/themes/detail?", "/api/popularity?"):
self.assertIn(endpoint, self.script)
screener_sources = "\n".join(
(
STATIC_DIR.parent
/ "backend"
/ "features"
/ "screener"
/ filename
).read_text(encoding="utf-8")
for filename in ("catalog.py", "factors.py")
)
for field in (
"auction_change", "auction_amount_million",
"auction_turnover_rate", "auction_volume_ratio",
):
self.assertIn(
field,
(
STATIC_DIR.parent
/ "backend"
/ "features"
/ "screener"
/ "engine.py"
).read_text(encoding="utf-8"),
)
self.assertIn(field, screener_sources)
def test_wencai_workspace_is_not_exposed_and_mentor_hides_internal_quality_score(self):
self.assertNotIn('id="wencaiView"', self.html)
@@ -160,23 +172,21 @@ class FrontendContractTests(unittest.TestCase):
self.assertIn("必需数据已完整,本日没有股票同时满足", self.script)
def test_dialogs_and_dark_table_hover_have_shared_safety_constraints(self):
redesign = (STATIC_DIR / "styles" / "redesign-v2.css").read_text(encoding="utf-8")
self.assertIn(".settings-dialog:not(.heaven-reading-dialog)[open] { margin: auto; }", redesign)
self.assertIn("max-height: min(760px, calc(100dvh - 28px));", redesign)
self.assertIn('#reviewWorkspaceView .data-table tbody tr:hover td', self.theme)
self.assertIn('#reviewWorkspaceView .data-table tbody td', self.theme)
self.assertIn('#screenerView .screener-result-frame tbody tr:hover td:last-child', self.theme)
self.assertIn(".settings-dialog:not(.heaven-reading-dialog)[open] {", self.auth_styles)
self.assertIn("margin: auto;", self.auth_styles)
self.assertIn("max-height: min(760px, -28px + 100dvh);", self.auth_styles)
self.assertIn('#reviewWorkspaceView .data-table tbody tr:hover td', self.review_styles)
self.assertIn('#reviewWorkspaceView .data-table tbody td', self.review_styles)
self.assertIn('#screenerView .screener-result-frame tbody tr:hover td:last-child', self.screener_styles)
def test_global_toast_has_one_owner_and_cannot_stretch_between_insets(self):
styles = (STATIC_DIR / "styles" / "styles.css").read_text(encoding="utf-8")
wentian = (STATIC_DIR / "pages" / "heaven" / "page.css").read_text(encoding="utf-8")
self.assertIn("#toast.toast {", styles)
self.assertIn("top: auto;", styles)
self.assertIn("left: auto;", styles)
self.assertIn("height: auto;", styles)
self.assertIn("#toast.toast[hidden] { display: none; }", styles)
self.assertNotIn(".toast{position:fixed", self.design_system)
self.assertNotRegex(wentian, r"(?m)^\.toast\s*\{")
self.assertIn("#toast.toast {", self.feedback_styles)
self.assertIn("inset: auto 18px 48px auto;", self.feedback_styles)
self.assertIn("width: max-content;", self.feedback_styles)
self.assertIn("height: auto;", self.feedback_styles)
self.assertIn("#toast.toast[hidden] {", self.feedback_styles)
self.assertIn("display: none;", self.feedback_styles)
self.assertNotRegex(self.heaven_styles, r"(?m)^\.toast\s*\{")
def test_public_knowledge_editors_are_hidden_for_non_admins(self):
self.assertIn('document.querySelector("#reasonForm").hidden = !isAdmin;', self.script)
@@ -185,8 +195,8 @@ class FrontendContractTests(unittest.TestCase):
def test_shared_ui_core_loads_before_application(self):
self.assertLess(
self.html.index('<script src="/shared/ui-core.js"'),
self.html.index('<script src="/app.js'),
self.registry.index('"/shared/ui-core.js"'),
self.registry.index('"/app.js'),
)
for function_name in (
"number", "clamp", "escapeHtml", "formatNumber", "formatTimestamp",
@@ -270,19 +280,19 @@ class FrontendContractTests(unittest.TestCase):
self.assertNotIn('context.fillStyle = "#6c7983";', chart)
def test_dark_mentor_tokens_and_sentiment_bottom_clearance_are_defined(self):
self.assertIn(":root[data-theme=\"dark\"] #mentorView {", self.theme)
self.assertIn("--mentor-ink: var(--text-primary);", self.theme)
self.assertIn("--mentor-sub: var(--text-secondary);", self.theme)
self.assertIn(":root[data-theme=\"dark\"] #mentorView .mentor-message {", self.theme)
self.assertIn("border-color: var(--line-soft);", self.theme)
self.assertIn("background: var(--surface-subtle);", self.theme)
self.assertIn("box-shadow: none;", self.theme)
self.assertIn("#sentimentCycleView .sentiment-history-frame {", self.theme)
self.assertIn("margin-bottom: var(--card-gap);", self.theme)
self.assertIn("padding-bottom: var(--card-gap);", self.theme)
self.assertIn(":root[data-theme=\"dark\"] #mentorView {", self.mentor_styles)
self.assertIn("--mentor-ink: var(--text-primary);", self.mentor_styles)
self.assertIn("--mentor-sub: var(--text-secondary);", self.mentor_styles)
self.assertIn(":root[data-theme=\"dark\"] #mentorView .mentor-message {", self.mentor_styles)
self.assertIn("border-color: var(--line-soft);", self.mentor_styles)
self.assertIn("background: var(--surface-subtle);", self.mentor_styles)
self.assertIn("box-shadow: none;", self.mentor_styles)
self.assertIn("#sentimentCycleView .sentiment-history-frame {", self.sentiment_styles)
self.assertIn("margin-bottom: var(--card-gap);", self.sentiment_styles)
self.assertIn("padding-bottom: var(--card-gap);", self.sentiment_styles)
self.assertIn("--sentiment-history-max-height: 510px;", self.tokens)
self.assertIn("max-height:var(--sentiment-history-max-height);", self.design_system)
self.assertIn("overflow:auto;", self.design_system)
self.assertIn("max-height: var(--sentiment-history-max-height);", self.sentiment_styles)
self.assertIn("overflow: auto;", self.sentiment_styles)
def test_theme_switch_is_atomic_and_theme_library_loading_surface_is_dark_safe(self):
self.assertIn('typeof document.startViewTransition === "function"', self.script)
@@ -290,9 +300,9 @@ class FrontendContractTests(unittest.TestCase):
self.assertIn('root.classList.remove("theme-switching")', self.script)
self.assertIn("clearThemeTransitionEffects();", self.script)
self.assertIn("redrawThemeSensitiveVisuals();", self.script)
self.assertIn(":root.theme-switching *", self.theme)
self.assertIn("::view-transition-old(root)", self.theme)
self.assertIn(".theme-detail-empty-v2,", self.theme)
self.assertIn(":root.theme-switching *", self.theme_library_styles)
self.assertIn("::view-transition-old(root)", self.base_styles)
self.assertIn(".theme-detail-empty-v2", self.theme_library_styles)
def test_membership_copy_includes_review_assistant_access(self):
self.assertIn("复盘助手仅对会员开放", self.html)
+6 -3
View File
@@ -4,7 +4,10 @@ import unittest
from pathlib import Path
from server import DashboardService
from tests.preservation_helpers import reassembled_frontend_runtime
from tests.frontend_test_helpers import (
assembled_frontend_runtime,
assembled_frontend_document,
)
class SearchDatabaseStub:
@@ -79,8 +82,8 @@ class GlobalSearchTests(unittest.TestCase):
def test_frontend_reuses_full_stock_detail_and_renders_market_daily_k(self):
static_dir = Path(__file__).resolve().parents[1] / "frontend"
html = (static_dir / "index.html").read_text(encoding="utf-8")
script = reassembled_frontend_runtime()
html = assembled_frontend_document()
script = assembled_frontend_runtime()
self.assertIn('id="globalSearchButton"', html)
self.assertIn('id="globalSearchDialog"', html)
+5 -5
View File
@@ -5,10 +5,10 @@ import json
import unittest
from unittest.mock import MagicMock, patch
from heaven_engine import _market_line_scores, build_manual_market_hexagram
from realtime_aggregator import WebRealtimeAggregator
from backend.data.realtime import WebRealtimeAggregator
from backend.features.heaven.engine import _market_line_scores, build_manual_market_hexagram
from server import DashboardService
from tushare_client import (
from backend.data.providers.tushare_client import (
TushareClient,
_filter_members_by_listing,
_sector_coverage_issue,
@@ -307,7 +307,7 @@ class RealtimeAggregatorTests(unittest.TestCase):
context.__enter__.return_value = response
return context
@patch("realtime_aggregator.urllib.request.urlopen")
@patch("backend.data.realtime.urllib.request.urlopen")
def test_transport_failure_is_retried(self, urlopen: MagicMock):
urlopen.side_effect = [
http.client.RemoteDisconnected("temporary disconnect"),
@@ -320,7 +320,7 @@ class RealtimeAggregatorTests(unittest.TestCase):
self.assertEqual(payload["rc"], 0)
self.assertEqual(urlopen.call_count, 2)
@patch("realtime_aggregator.urllib.request.urlopen")
@patch("backend.data.realtime.urllib.request.urlopen")
def test_recent_success_is_used_after_retries_fail(self, urlopen: MagicMock):
aggregator = WebRealtimeAggregator(retry_delay_seconds=0)
urlopen.return_value = self._response({"rc": 0, "data": {"diff": []}})
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from tushare_client import TushareClient
from backend.data.providers.tushare_client import TushareClient
class HotMoneyProfileClient(TushareClient):
+62
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
import ast
import unittest
from pathlib import Path
from backend.application import (
AUTHENTICATED_POST_HANDLERS,
@@ -9,6 +11,23 @@ from backend.application import (
)
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:
@@ -78,6 +97,49 @@ class HttpDispatchContractTests(unittest.TestCase):
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()
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from ifind_client import IfindError, IfindHttpClient
from backend.data.providers.ifind_client import IfindError, IfindHttpClient
class IfindClientTests(unittest.TestCase):
+3 -3
View File
@@ -6,9 +6,9 @@ from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
from chart_data_provider import EastmoneyChartClient, MarketChartClient
from backend.features.market.charts import EastmoneyChartClient, MarketChartClient
from database import ReviewDatabase
from market_insights import MarketInsightsService
from backend.features.market.insights import MarketInsightsService
from server import DashboardService
@@ -135,7 +135,7 @@ class IfindFeatureTests(unittest.TestCase):
def test_ifind_daily_chart_keeps_last_traded_bar_before_market_open(self):
client = MarketChartClient(FakeIfindStalePreopen(), EastmoneyChartClient())
with patch("chart_data_provider.datetime", FixedPreopenDatetime):
with patch("backend.features.market.charts.datetime", FixedPreopenDatetime):
rows = client.stock_daily("000001", "20260729")
self.assertEqual(rows[-1]["trade_date"], "2026-07-28")
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from llm_stream import OpenAIStreamAccumulator
from backend.llm.stream import OpenAIStreamAccumulator
class OpenAIStreamAccumulatorTests(unittest.TestCase):
+16 -8
View File
@@ -7,11 +7,11 @@ import urllib.error
from pathlib import Path
from unittest.mock import patch
from assistant_agent import ReviewAssistantError, stream_review_assistant
from backend.features.heaven.agent import HeavenAgentError, interpret_heaven
from backend.features.mentor.agent import MentorAgentError, MentorSkill, stream_with_mentor
from backend.features.review.agent import ReviewAssistantError, stream_review_assistant
from backend.features.screener.compiler import LLMCompilerError, test_llm_connection
from backend.llm import transport
from heaven_agent import HeavenAgentError, interpret_heaven
from llm_strategy import LLMCompilerError, test_llm_connection
from mentor_agent import MentorAgentError, MentorSkill, stream_with_mentor
ROOT = Path(__file__).resolve().parents[1]
@@ -158,25 +158,33 @@ class FeatureErrorMappingTests(unittest.TestCase):
path=Path("SKILL.md"),
)
with patch(
"mentor_agent.llm_transport.stream_chat_completion", side_effect=error
"backend.features.mentor.agent.llm_transport.stream_chat_completion",
side_effect=error,
):
with self.assertRaisesRegex(
MentorAgentError, "问师模型调用失败(HTTP 429):capacity"
):
list(stream_with_mentor(skill, {}, "问题", [], "key", "https://x", "m"))
with patch("heaven_agent.llm_transport.chat_completion", side_effect=error):
with patch(
"backend.features.heaven.agent.llm_transport.chat_completion",
side_effect=error,
):
with self.assertRaisesRegex(
HeavenAgentError, "问天模型调用失败(HTTP 429):capacity"
):
interpret_heaven("heart", {}, "key", "https://x", "m")
with patch(
"assistant_agent.llm_transport.stream_chat_completion", side_effect=error
"backend.features.review.agent.llm_transport.stream_chat_completion",
side_effect=error,
):
with self.assertRaisesRegex(
ReviewAssistantError, "智能解读服务暂不可用(429"
):
list(stream_review_assistant({}, "问题", [], "key", "https://x", "m"))
with patch("llm_strategy.llm_transport.chat_completion", side_effect=error):
with patch(
"backend.features.screener.compiler.llm_transport.chat_completion",
side_effect=error,
):
with self.assertRaisesRegex(
LLMCompilerError, "模型连接测试失败(HTTP 429):capacity"
):
+24 -18
View File
@@ -19,7 +19,7 @@ ROOT = Path(__file__).resolve().parents[1]
class MaintenanceToolTests(unittest.TestCase):
def test_generated_candidate_registries_are_current(self) -> None:
def test_generated_registries_are_current(self) -> None:
expected_api = json.loads(
(ROOT / "config" / "api.config.json").read_text(encoding="utf-8")
)
@@ -31,20 +31,29 @@ class MaintenanceToolTests(unittest.TestCase):
self.assertEqual(expected_api, build_api_registry())
self.assertEqual(expected_architecture, build_architecture_inventory())
def test_baseline_verifier_uses_the_candidate_frontend(self) -> None:
def test_baseline_verifier_uses_the_application_frontend(self) -> None:
source = (ROOT / "tools" / "verify_baseline.py").read_text(encoding="utf-8")
self.assertIn('FRONTEND_ROOT = ROOT / "frontend"', source)
self.assertIn('FRONTEND_ROOT.rglob("*.js")', source)
self.assertIn('path.suffix in {".js", ".mjs"}', source)
self.assertIn("start_e2e_server()", source)
self.assertIn("stop_e2e_server(server)", source)
self.assertNotIn('"static/app.js"', source)
def test_standalone_verifier_excludes_only_migration_comparison_modules(self) -> None:
repository_command = python_test_command(preservation_baseline=True)
standalone_command = python_test_command(preservation_baseline=False)
self.assertIn("discover", repository_command)
self.assertTrue(any(item == "tests.test_frontend_contract" for item in standalone_command))
self.assertFalse(any("test_preservation_" in item for item in standalone_command))
def test_runtime_artifacts_have_one_ignored_root(self) -> None:
gitignore = (ROOT / ".gitignore").read_text(encoding="utf-8")
dockerignore = (ROOT / ".dockerignore").read_text(encoding="utf-8")
playwright = (ROOT / "playwright.config.js").read_text(encoding="utf-8")
launcher = (ROOT / "tools" / "start_local.ps1").read_text(encoding="utf-8")
self.assertIn("runtime/*", gitignore)
self.assertIn("runtime/", dockerignore)
self.assertIn('outputDir: "./runtime/test-results"', playwright)
self.assertIn('Join-Path $runtime "logs"', launcher)
self.assertEqual(list(ROOT.glob("*.log")), [])
def test_verifier_runs_the_complete_standalone_suite(self) -> None:
command = python_test_command()
self.assertIn("discover", command)
self.assertIn("tests", command)
def test_architecture_metrics_are_independent_of_checkout_line_endings(self) -> None:
with tempfile.TemporaryDirectory() as directory:
@@ -70,19 +79,16 @@ class MaintenanceToolTests(unittest.TestCase):
)
self.assertEqual(result.returncode, 0, result.stderr)
def test_migration_only_tools_are_explicitly_classified(self) -> None:
readme = (ROOT / "tools" / "README.md").read_text(encoding="utf-8")
def test_migration_only_tools_are_retired(self) -> None:
for name in (
"build_preservation_manifest.py",
"compare_preservation_apis.py",
"compare_preservation_databases.py",
"move_class_methods.py",
"split_frontend_runtime.py",
"run_preservation_runtime.py",
):
self.assertIn(name, readme)
manifest = (ROOT / "tools" / "build_preservation_manifest.py").read_text(
encoding="utf-8"
)
self.assertNotIn('TARGET = ROOT / "app"', manifest)
self.assertIn('required=True', manifest)
with self.subTest(name=name):
self.assertFalse((ROOT / "tools" / name).exists())
if __name__ == "__main__":
+3 -3
View File
@@ -6,9 +6,9 @@ from datetime import datetime, timedelta, timezone
from pathlib import Path
from database import ReviewDatabase
from market_insights import MarketInsightsService
from screener import FACTOR_FIELDS, ScreenerEngine
from tushare_client import TushareError
from backend.data.providers.tushare_client import TushareError
from backend.features.market.insights import MarketInsightsService
from backend.features.screener.engine import FACTOR_FIELDS, ScreenerEngine
class FakeMarketClient:
+3 -3
View File
@@ -5,15 +5,15 @@ import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from tushare_client import _sector_coverage_issue
from backend.data.providers.tushare_client import _sector_coverage_issue
def load_method(name: str):
source = Path("backend/features/heaven/service.py").read_text(encoding="utf-8")
source = Path("backend/features/heaven/trend.py").read_text(encoding="utf-8")
tree = ast.parse(source)
dashboard_service = next(
node for node in tree.body
if isinstance(node, ast.ClassDef) and node.name == "HeavenServiceMixin"
if isinstance(node, ast.ClassDef) and node.name == "HeavenTrendMixin"
)
method = next(
node for node in dashboard_service.body
+1 -1
View File
@@ -5,7 +5,7 @@ import tempfile
import unittest
from pathlib import Path
from mentor_agent import MentorSkillRegistry
from backend.features.mentor.agent import MentorSkillRegistry
ROOT = Path(__file__).resolve().parents[1]
+1 -1
View File
@@ -5,7 +5,7 @@ import unittest
from pathlib import Path
from unittest.mock import patch
from mentor_agent import MentorSkill, chat_with_mentor, stream_with_mentor
from backend.features.mentor.agent import MentorSkill, chat_with_mentor, stream_with_mentor
class FakeStreamResponse:
-118
View File
@@ -1,118 +0,0 @@
from __future__ import annotations
import tempfile
import threading
import unittest
from pathlib import Path
import server
from backend.application import DashboardService, RequestHandler
from backend.bootstrap.config import APP_DIR, STATIC_DIR
from backend.features.accounts.repository import AccountRepositoryMixin
from backend.features.accounts.security import SecretVault, token_hash
from backend.features.accounts.service import AccountService
from backend.http.handler import HttpTransportMixin
from database import ReviewDatabase
class AccountSliceStructureTests(unittest.TestCase):
def test_original_entrypoint_exports_canonical_runtime(self) -> None:
self.assertIs(server.DashboardService, DashboardService)
self.assertIs(server.RequestHandler, RequestHandler)
self.assertIs(server.SERVICE, RequestHandler.application_service)
def test_runtime_paths_still_point_at_app_root(self) -> None:
self.assertEqual(APP_DIR, Path(__file__).resolve().parents[1])
self.assertEqual(STATIC_DIR, APP_DIR / "frontend")
def test_account_persistence_and_http_transport_have_single_owners(self) -> None:
for method in (
"create_user",
"session_user",
"update_membership",
"save_user_birth_profile",
):
self.assertNotIn(method, ReviewDatabase.__dict__)
self.assertIn(method, AccountRepositoryMixin.__dict__)
for method in (
"require_auth",
"require_csrf",
"require_access",
"serve_static",
"send_json",
"_write_stream_event",
"send_ndjson_stream",
):
self.assertNotIn(method, RequestHandler.__dict__)
self.assertIn(method, HttpTransportMixin.__dict__)
class AccountSliceBehaviorTests(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.database = ReviewDatabase(Path(self.temporary.name) / "review.db")
self.vault = SecretVault(SecretVault.generate_key())
self.context: dict[str, object] = {"user_id": 0, "access": {}}
def bind_user(user_id: int) -> None:
self.context["user_id"] = user_id
self.context["access"] = self.database.user_access(user_id) or {}
self.service = AccountService(
database=self.database,
vault=self.vault,
current_user_supplier=lambda: int(self.context["user_id"]),
access_supplier=lambda: dict(self.context["access"]),
bind_user=bind_user,
personal_field_builder=lambda *args: {
"birth": "private",
"day_master": "甲木",
"current": {"trade_date": args[2]},
"notice": "test",
},
auth_lock=threading.Lock(),
)
def tearDown(self) -> None:
self.temporary.cleanup()
def test_register_login_session_and_password_contract(self) -> None:
registered = self.service.register("owner_01", "Password123")
self.assertEqual(registered["user"]["role"], "admin")
self.assertTrue(registered["user"]["membership"]["active"])
self.assertIsNotNone(
self.database.session_user(token_hash(registered["session_token"]))
)
with self.assertRaisesRegex(ValueError, "账号名或密码不正确"):
self.service.login("owner_01", "wrong-password")
self.service.change_password("Password123", "NewPassword456")
logged_in = self.service.login("owner_01", "NewPassword456")
self.assertEqual(logged_in["user"]["id"], registered["user"]["id"])
def test_membership_and_birth_profile_remain_account_scoped(self) -> None:
owner = self.service.register("owner_02", "Password123")
other = self.database.create_user("other_02", "salt", "hash")
self.service.update_membership(
{"user_id": other["id"], "status": "active", "duration": "3_months"}
)
other_access = self.database.user_access(other["id"])
self.assertEqual(other_access["membership_plan"], "3个月")
self.assertTrue(AccountService.membership_for_access(other_access)["subscribed"])
personal = self.service.save_birth_profile(
{
"birth_datetime": "1990-01-01 08:30",
"gender": "male",
"trade_date": "2026-07-30",
}
)
self.assertNotIn("birth", personal)
self.assertEqual(personal["day_master"], "甲木")
self.assertTrue(self.database.get_user_birth_profile(owner["user"]["id"]))
self.assertEqual(self.database.get_user_birth_profile(other["id"]), "")
if __name__ == "__main__":
unittest.main()
-96
View File
@@ -1,96 +0,0 @@
from __future__ import annotations
import re
import unittest
from backend.bootstrap.config import APP_DIR, STATIC_DIR
from tests.preservation_helpers import (
FRONTEND_ROOT,
ORIGINAL_STATIC,
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
)
class FrontendPreservationSliceTests(unittest.TestCase):
def test_split_runtime_matches_original_except_audited_retirements(self) -> None:
assert_frontend_runtime_matches_audited_baseline(self)
def test_index_diff_is_limited_to_asset_relocation_and_split_loading(self) -> None:
migrated = (FRONTEND_ROOT / "index.html").read_text(encoding="utf-8")
restored = migrated
for current, original in (
("/styles/styles.css", "/styles.css"),
("/styles/renovation.css", "/renovation.css"),
("/styles/redesign-v2.css", "/redesign-v2.css"),
("/styles/design-system.css", "/design-system.css"),
("/styles/theme.css", "/theme.css"),
("/pages/heaven/page.css", "/wentian-v2.css"),
("/shared/ui-core.js", "/ui-core.js"),
("/pages/heaven/loading-v2.js", "/heaven-loading-v2.js"),
):
restored = restored.replace(current, original)
restored = restored.replace(
' <script src="/pages/market/runtime.js?v=20260731-1" defer></script>\n',
"",
)
restored = restored.replace(
' <script src="/shared/export.js?v=20260731-1" defer></script>\n',
"",
)
self.assertEqual(
restored,
(ORIGINAL_STATIC / "index.html").read_text(encoding="utf-8"),
)
def test_stylesheet_stack_matches_baseline_after_audited_retirements(self) -> None:
for original, migrated in (
("shared/tokens.css", "shared/tokens.css"),
("styles.css", "styles/styles.css"),
("renovation.css", "styles/renovation.css"),
("redesign-v2.css", "styles/redesign-v2.css"),
("design-system.css", "styles/design-system.css"),
("theme.css", "styles/theme.css"),
("wentian-v2.css", "pages/heaven/page.css"),
):
with self.subTest(asset=original):
assert_moved_asset_matches(self, original, migrated)
def test_shared_vendor_and_animation_assets_are_byte_identical(self) -> None:
for original, migrated in (
("ui-core.js", "shared/ui-core.js"),
("heaven-loading-v2.js", "pages/heaven/loading-v2.js"),
("vendor/lucide.min.js", "vendor/lucide.min.js"),
("pages.config.js", "pages.config.js"),
("pages/runtime.js", "pages/runtime.js"),
("shared/api.js", "shared/api.js"),
("shared/components.js", "shared/components.js"),
("shared/shell.js", "shared/shell.js"),
("shared/state.js", "shared/state.js"),
):
with self.subTest(asset=original):
assert_moved_asset_matches(self, original, migrated)
def test_original_page_registration_prefixes_are_preserved(self) -> None:
for path in sorted((ORIGINAL_STATIC / "pages").glob("*/page.js")):
relative = path.relative_to(ORIGINAL_STATIC).as_posix()
with self.subTest(page=relative):
assert_page_prefix_matches(self, relative)
def test_frontend_is_the_only_served_static_root(self) -> None:
self.assertEqual(STATIC_DIR, APP_DIR / "frontend")
self.assertFalse((APP_DIR / "static").exists())
def test_shared_api_remains_the_only_fetch_exit(self) -> None:
consumers = []
for path in FRONTEND_ROOT.rglob("*.js"):
if "vendor" in path.parts:
continue
if re.search(r"\bfetch\s*\(", path.read_text(encoding="utf-8")):
consumers.append(path.relative_to(FRONTEND_ROOT).as_posix())
self.assertEqual(consumers, ["shared/api.js"])
if __name__ == "__main__":
unittest.main()
-167
View File
@@ -1,167 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import heaven_agent
import heaven_engine
from backend.features.heaven import agent as canonical_agent
from backend.features.heaven import engine as canonical_engine
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
HEAVEN_SERVICE_METHODS = {
"_heaven_manual_schema",
"_validate_heaven_manual_data",
"_apply_heaven_manual_data",
"_heaven_line_checks",
"_resolve_heaven_stock_code",
"heaven_setup",
"_heaven_stock_context",
"_heaven_market_mode",
"_heaven_trend_sources",
"_heaven_trend_quality_issues",
"heaven_personal",
"heaven_hexagram",
"heaven_readings",
"_heaven_reading_identity",
"heaven_interpret",
"_legacy_truncated_heaven_reading",
"_call_heaven_agent",
"_heaven_index_context",
"_aggregate_index_context",
"_heaven_sector_context",
}
HEAVEN_REPOSITORY_METHODS = {
"list_sector_phase_overrides",
"save_sector_phase_override",
"delete_sector_phase_override",
"_heaven_reading_dict",
"save_heaven_reading",
"list_heaven_readings",
"latest_heaven_reading",
"delete_heaven_reading",
}
HEAVEN_HTTP_METHODS = {"heaven_hexagram", "heaven_personal", "heaven_interpret"}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def top_level_definitions(path: Path) -> dict[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
return {
node.name: ast.dump(node, include_attributes=False)
for node in tree.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
}
class HeavenSliceSourceEquivalenceTests(unittest.TestCase):
def assert_methods_equal(
self,
original_path: Path,
original_class: str,
migrated_path: Path,
migrated_class: str,
expected: set[str],
adapted: set[str] | None = None,
) -> None:
original = class_methods(original_path, original_class)
migrated = class_methods(migrated_path, migrated_class)
self.assertEqual(set(migrated), expected)
for name in sorted(expected - (adapted or set())):
self.assertEqual(migrated[name], original[name], name)
def test_heaven_agent_uses_shared_transport(self) -> None:
source = (
APP_ROOT / "backend" / "features" / "heaven" / "agent.py"
).read_text(encoding="utf-8")
self.assertIn("llm_transport.chat_completion", source)
self.assertNotIn("urllib.request", source)
def test_heaven_engine_definitions_are_exact_original_ast(self) -> None:
self.assertEqual(
top_level_definitions(ORIGINAL_ROOT / "heaven_engine.py"),
top_level_definitions(
APP_ROOT / "backend" / "features" / "heaven" / "engine.py"
),
)
def test_compatibility_modules_are_canonical_module_objects(self) -> None:
self.assertIs(heaven_agent, canonical_agent)
self.assertIs(heaven_engine, canonical_engine)
def test_heaven_service_methods_are_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "server.py",
"DashboardService",
APP_ROOT / "backend" / "features" / "heaven" / "service.py",
"HeavenServiceMixin",
HEAVEN_SERVICE_METHODS,
{"_heaven_reading_identity"},
)
source = (
APP_ROOT / "backend" / "features" / "heaven" / "service.py"
).read_text(encoding="utf-8")
self.assertIn(
"MarketServiceMixin._display_compact_date(context_date)", source
)
def test_heaven_repository_methods_are_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "database.py",
"ReviewDatabase",
APP_ROOT / "backend" / "features" / "heaven" / "repository.py",
"HeavenRepositoryMixin",
HEAVEN_REPOSITORY_METHODS,
)
def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
remaining_service = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
remaining_http = class_methods(
APP_ROOT / "backend" / "application.py", "RequestHandler"
)
self.assertTrue(HEAVEN_SERVICE_METHODS.isdisjoint(remaining_service))
self.assertTrue(HEAVEN_REPOSITORY_METHODS.isdisjoint(remaining_database))
self.assertTrue(HEAVEN_HTTP_METHODS.isdisjoint(remaining_http))
def test_http_mixin_preserves_all_heaven_endpoints(self) -> None:
methods = class_methods(
APP_ROOT / "backend" / "features" / "heaven" / "http.py",
"HeavenHttpMixin",
)
self.assertEqual(set(methods), HEAVEN_HTTP_METHODS)
source = (
APP_ROOT / "backend" / "features" / "heaven" / "http.py"
).read_text(encoding="utf-8")
self.assertNotIn("SERVICE.", source)
self.assertEqual(source.count("self.application_service.heaven_"), 3)
if __name__ == "__main__":
unittest.main()
@@ -1,93 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
)
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
ROTATION_METHODS = {
"rotation_history",
"rotation_sector_members",
}
LADDER_ROTATION_BUILDERS = {
"_build_ladders",
"_build_sector_rotation",
}
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def top_level_functions(path: Path) -> dict[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
return {
node.name: ast.dump(node, include_attributes=False)
for node in tree.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name in LADDER_ROTATION_BUILDERS
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
class LadderRotationSliceSourceEquivalenceTests(unittest.TestCase):
def test_rotation_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "rotation" / "service.py",
"RotationServiceMixin",
)
self.assertEqual(set(migrated), ROTATION_METHODS)
for name in sorted(ROTATION_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_dashboard_service_no_longer_duplicates_rotation_methods(self) -> None:
remaining = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
self.assertTrue(ROTATION_METHODS.isdisjoint(remaining))
def test_ladder_and_rotation_builders_are_exact_original_ast(self) -> None:
self.assertEqual(
top_level_functions(ORIGINAL_ROOT / "tushare_client.py"),
top_level_functions(
APP_ROOT / "backend" / "data" / "providers" / "tushare_client.py"
),
)
def test_api_and_frontend_assets_are_unchanged(self) -> None:
self.assertEqual(
sha256(APP_ROOT / "config/api.config.json"),
sha256(ORIGINAL_ROOT / "config/api.config.json"),
)
assert_frontend_runtime_matches_audited_baseline(self)
assert_moved_asset_matches(self, "styles.css", "styles/styles.css")
for page in ("pages/ladder/page.js", "pages/rotation/page.js"):
assert_page_prefix_matches(self, page)
if __name__ == "__main__":
unittest.main()
-195
View File
@@ -1,195 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import chart_data_provider
import ifind_client
import realtime_aggregator
import tushare_client
from backend.bootstrap import config as bootstrap_config
from backend.data import realtime
from backend.data.numbers import finite_number
from backend.data.providers import ifind_client as canonical_ifind
from backend.data.providers import tushare_client as canonical_tushare
from backend.features.market import charts
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
function_contract,
)
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
MARKET_METHODS = {
"_tushare_client",
"_market_insights",
"get_dashboard",
"_dashboard_sentiment_ready",
"_display_compact_date",
"_carry_dashboard",
"_realtime_snapshot_due",
"sync_dashboard",
"realtime_aggregate_health",
"_search_market_directory",
"_search_match_score",
"search_entities",
"get_search_detail",
"get_intraday_chart",
"_ths_search_detail",
"_index_search_detail",
"get_stock_detail",
"_stock_detail_bar_date",
"_stock_detail_cache_needs_refresh",
"_prepare_stock_detail",
"_sanitize_stock_detail_prices",
"_valid_realtime_stock_quote",
"_ifind_realtime_stock_quote",
"_merge_realtime_stock_detail",
"get_stock_preview",
"backfill",
"_stock_identity",
"_enrich_stock_detail",
"_with_storage",
"_record_count",
}
MARKET_REPOSITORY_METHODS = {
"get_snapshot",
"get_latest_real_snapshot",
"save_snapshot",
"get_data_snapshot",
"get_latest_data_snapshot",
"save_data_snapshot",
"search_stock_master",
"list_snapshot_payloads",
"start_sync",
"finish_sync",
"status",
"upsert_stock_master",
"list_stock_master",
"upsert_daily_bars",
"daily_bars_for_date",
}
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def top_level_definitions(path: Path) -> dict[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
return {
node.name: ast.dump(node, include_attributes=False)
for node in tree.body
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef))
}
class MarketSliceSourceEquivalenceTests(unittest.TestCase):
def test_market_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "market" / "service.py",
"MarketServiceMixin",
)
self.assertEqual(set(migrated), MARKET_METHODS)
for name in sorted(MARKET_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_market_repository_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "database.py", "ReviewDatabase")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "market" / "repository.py",
"MarketRepositoryMixin",
)
self.assertEqual(set(migrated), MARKET_REPOSITORY_METHODS)
for name in sorted(MARKET_REPOSITORY_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
remaining_service = class_methods(APP_ROOT / "backend" / "application.py", "DashboardService")
remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
self.assertTrue(MARKET_METHODS.isdisjoint(remaining_service))
self.assertTrue(MARKET_REPOSITORY_METHODS.isdisjoint(remaining_database))
def test_provider_compatibility_modules_are_canonical_aliases(self) -> None:
self.assertIs(tushare_client.TushareClient, canonical_tushare.TushareClient)
self.assertIs(ifind_client.IfindHttpClient, canonical_ifind.IfindHttpClient)
self.assertIs(realtime_aggregator.WebRealtimeAggregator, realtime.WebRealtimeAggregator)
self.assertIs(chart_data_provider.MarketChartClient, charts.MarketChartClient)
def test_provider_logic_is_the_original_implementation(self) -> None:
exact_moves = (
("ifind_client.py", "backend/data/providers/ifind_client.py"),
("realtime_aggregator.py", "backend/data/realtime.py"),
)
for original, migrated in exact_moves:
self.assertEqual(sha256(ORIGINAL_ROOT / original), sha256(APP_ROOT / migrated))
original_tushare = top_level_definitions(ORIGINAL_ROOT / "tushare_client.py")
original_tushare.pop("_number")
original_tushare.pop("_display_date")
self.assertEqual(
original_tushare,
top_level_definitions(APP_ROOT / "backend/data/providers/tushare_client.py"),
)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "tushare_client.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "finite_number"),
)
self.assertIs(canonical_tushare._number, finite_number)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "tushare_client.py", "_display_date"),
function_contract(
APP_ROOT / "backend/bootstrap/config.py", "display_compact_date"
),
)
self.assertIs(canonical_tushare._display_date, bootstrap_config.display_compact_date)
original_charts = top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py")
original_charts.pop("_stock_market_code")
self.assertEqual(
original_charts,
top_level_definitions(APP_ROOT / "backend/features/market/charts.py"),
)
self.assertEqual(
function_contract(
ORIGINAL_ROOT / "chart_data_provider.py", "_stock_market_code"
),
function_contract(
APP_ROOT / "backend/bootstrap/config.py", "tushare_code"
),
)
self.assertIs(charts._stock_market_code, bootstrap_config.tushare_code)
def test_relocated_frontend_preserves_original_market_runtime_and_styles(self) -> None:
assert_frontend_runtime_matches_audited_baseline(self)
for original, migrated in (
("styles.css", "styles/styles.css"),
("renovation.css", "styles/renovation.css"),
("redesign-v2.css", "styles/redesign-v2.css"),
("theme.css", "styles/theme.css"),
("wentian-v2.css", "pages/heaven/page.css"),
):
assert_moved_asset_matches(self, original, migrated)
if __name__ == "__main__":
unittest.main()
@@ -1,205 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import market_insights
from backend.data.numbers import non_nan_number
from backend.features.market import insights as canonical_insights
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
function_contract,
)
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
MARKET_INSIGHT_METHODS = {
"__init__",
"_trade_context",
"_latest_feature_snapshot",
"_auction_session",
"_stock_master",
"_expectation_label",
"_auction_confirmation",
"_attention_score",
"_auction_candidates",
"_auction_theme_evidence",
"_auction_amount_history",
"_ensure_auction_amount_history",
"_with_auction_watchlist",
"_dynamic_auction_rows",
"auction_center",
"_theme_directory",
"theme_library",
"theme_detail",
"_parse_concepts",
"popularity",
"_hot_rows",
"_normalize_hot",
}
MARKET_SERVICE_METHODS = {"_market_insights"}
AUCTION_SERVICE_METHODS = {"auction_center"}
THEME_SERVICE_METHODS = {"theme_library", "theme_detail"}
POPULARITY_SERVICE_METHODS = {"popularity"}
DRAGON_TIGER_SERVICE_METHODS = {
"get_hot_money_profiles",
"get_dragon_tiger",
"_apply_seat_aliases",
}
AUCTION_REPOSITORY_METHODS = {
"upsert_auction_factors",
"auction_factor_dates",
"auction_factors_for_date",
}
POPULARITY_REPOSITORY_METHODS = {"upsert_popularity_factors"}
DRAGON_TIGER_REPOSITORY_METHODS = {
"list_seat_aliases",
"save_seat_alias",
"upsert_lhb_institutions",
}
TUSHARE_METHODS = {"hot_money_profiles", "dragon_tiger"}
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
class MarketInsightsSliceSourceEquivalenceTests(unittest.TestCase):
def assert_methods_equal(
self,
original_path: Path,
original_class: str,
migrated_path: Path,
migrated_class: str,
names: set[str],
) -> None:
original = class_methods(original_path, original_class)
migrated = class_methods(migrated_path, migrated_class)
self.assertEqual(set(migrated), names)
for name in sorted(names):
self.assertEqual(migrated[name], original[name], name)
def test_shared_market_insight_service_is_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "market_insights.py",
"MarketInsightsService",
APP_ROOT / "backend" / "features" / "market" / "insights.py",
"MarketInsightsService",
MARKET_INSIGHT_METHODS,
)
self.assertIs(market_insights.MarketInsightsService, canonical_insights.MarketInsightsService)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "market_insights.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "non_nan_number"),
)
self.assertIs(canonical_insights._number, non_nan_number)
def test_dashboard_service_methods_are_exact_original_ast(self) -> None:
original = ORIGINAL_ROOT / "server.py"
mappings = (
("auction/service.py", "AuctionServiceMixin", AUCTION_SERVICE_METHODS),
("themes/service.py", "ThemeServiceMixin", THEME_SERVICE_METHODS),
("popularity/service.py", "PopularityServiceMixin", POPULARITY_SERVICE_METHODS),
("dragon_tiger/service.py", "DragonTigerServiceMixin", DRAGON_TIGER_SERVICE_METHODS),
)
for relative, class_name, names in mappings:
with self.subTest(relative=relative):
self.assert_methods_equal(
original,
"DashboardService",
APP_ROOT / "backend" / "features" / relative,
class_name,
names,
)
original_methods = class_methods(original, "DashboardService")
market_methods = class_methods(
APP_ROOT / "backend" / "features" / "market" / "service.py",
"MarketServiceMixin",
)
for name in MARKET_SERVICE_METHODS:
self.assertEqual(market_methods[name], original_methods[name], name)
def test_repository_methods_are_exact_original_ast(self) -> None:
original = ORIGINAL_ROOT / "database.py"
mappings = (
("auction/repository.py", "AuctionRepositoryMixin", AUCTION_REPOSITORY_METHODS),
("popularity/repository.py", "PopularityRepositoryMixin", POPULARITY_REPOSITORY_METHODS),
("dragon_tiger/repository.py", "DragonTigerRepositoryMixin", DRAGON_TIGER_REPOSITORY_METHODS),
)
for relative, class_name, names in mappings:
with self.subTest(relative=relative):
self.assert_methods_equal(
original,
"ReviewDatabase",
APP_ROOT / "backend" / "features" / relative,
class_name,
names,
)
def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
remaining_service = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
moved_service = (
MARKET_SERVICE_METHODS
| AUCTION_SERVICE_METHODS
| THEME_SERVICE_METHODS
| POPULARITY_SERVICE_METHODS
| DRAGON_TIGER_SERVICE_METHODS
)
moved_repository = (
AUCTION_REPOSITORY_METHODS
| POPULARITY_REPOSITORY_METHODS
| DRAGON_TIGER_REPOSITORY_METHODS
)
self.assertTrue(moved_service.isdisjoint(remaining_service))
self.assertTrue(moved_repository.isdisjoint(remaining_database))
def test_tushare_dragon_tiger_implementations_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "tushare_client.py", "TushareClient")
migrated = class_methods(
APP_ROOT / "backend" / "data" / "providers" / "tushare_client.py",
"TushareClient",
)
for name in sorted(TUSHARE_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_api_and_frontend_assets_are_unchanged(self) -> None:
self.assertEqual(
sha256(APP_ROOT / "config/api.config.json"),
sha256(ORIGINAL_ROOT / "config/api.config.json"),
)
assert_frontend_runtime_matches_audited_baseline(self)
assert_moved_asset_matches(self, "styles.css", "styles/styles.css")
for page in (
"pages/auction/page.js",
"pages/themes/page.js",
"pages/popularity/page.js",
"pages/dragon-tiger/page.js",
):
assert_page_prefix_matches(self, page)
if __name__ == "__main__":
unittest.main()
-205
View File
@@ -1,205 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import llm_stream
import mentor_agent
from backend.features.mentor import agent as canonical_agent
from backend.llm import stream as canonical_stream
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
MENTOR_SERVICE_METHODS = {
"mentor_setup",
"save_mentor_preferences",
"mentor_stream",
"mentor_messages",
"clear_mentor_messages",
"_validate_mentor_history",
"_build_mentor_context",
"_mentor_market_matrix",
}
LLM_SERVICE_METHODS = {
"_personal_llm_profile",
"_platform_llm_profile",
"_profile_configured",
"_resolved_llm_profile",
"llm_primary_api_key",
"llm_primary_base_url",
"llm_primary_model",
"llm_fallback_api_key",
"llm_fallback_base_url",
"llm_fallback_model",
"llm_source",
"llm_configured",
"llm_fallback_configured",
"save_llm_settings",
"save_llm_mode",
"test_llm_profile",
"_validate_llm_profile",
"llm_access_status",
"_platform_usage_today",
"_platform_usage_today_for_user",
"test_system_llm_profile",
}
MENTOR_REPOSITORY_METHODS = {
"save_mentor_exchange",
"list_mentor_messages",
"delete_mentor_messages",
"list_mentor_preferences",
"save_mentor_preferences",
}
LLM_REPOSITORY_METHODS = {"record_llm_usage", "count_llm_usage_since"}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def assignments(path: Path, names: set[str]) -> dict[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
result = {}
for node in tree.body:
if not isinstance(node, ast.Assign) or len(node.targets) != 1:
continue
target = node.targets[0]
if isinstance(target, ast.Name) and target.id in names:
result[target.id] = ast.dump(node.value, include_attributes=False)
return result
class MentorLLMSliceSourceEquivalenceTests(unittest.TestCase):
def assert_methods_equal(
self,
original_path: Path,
original_class: str,
migrated_path: Path,
migrated_class: str,
expected: set[str],
) -> None:
original = class_methods(original_path, original_class)
migrated = class_methods(migrated_path, migrated_class)
self.assertEqual(set(migrated), expected)
for name in sorted(expected):
self.assertEqual(migrated[name], original[name], name)
def test_mentor_agent_uses_shared_transport_and_stream_accumulator_is_exact(self) -> None:
mentor_source = (APP_ROOT / "backend" / "features" / "mentor" / "agent.py").read_text(
encoding="utf-8"
)
self.assertIn("llm_transport.stream_chat_completion", mentor_source)
self.assertNotIn("urllib.request", mentor_source)
self.assertEqual(
sha256(ORIGINAL_ROOT / "llm_stream.py"),
sha256(APP_ROOT / "backend" / "llm" / "stream.py"),
)
def test_compatibility_modules_are_canonical_module_objects(self) -> None:
self.assertIs(mentor_agent, canonical_agent)
self.assertIs(llm_stream, canonical_stream)
def test_mentor_service_methods_are_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "server.py",
"DashboardService",
APP_ROOT / "backend" / "features" / "mentor" / "service.py",
"MentorServiceMixin",
MENTOR_SERVICE_METHODS,
)
def test_llm_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
migrated = class_methods(
APP_ROOT / "backend" / "llm" / "service.py", "LLMServiceMixin"
)
self.assertEqual(set(migrated), LLM_SERVICE_METHODS)
adapted = {"_platform_usage_today", "_platform_usage_today_for_user"}
for name in sorted(LLM_SERVICE_METHODS - adapted):
self.assertEqual(migrated[name], original[name], name)
source = (APP_ROOT / "backend" / "llm" / "service.py").read_text(
encoding="utf-8"
)
self.assertIn(
"return self._platform_usage_today_for_user(self.current_user_id)", source
)
self.assertIn("def _platform_usage_today_for_user(self, user_id: int)", source)
def test_mentor_and_llm_repositories_are_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "database.py",
"ReviewDatabase",
APP_ROOT / "backend" / "features" / "mentor" / "repository.py",
"MentorRepositoryMixin",
MENTOR_REPOSITORY_METHODS,
)
self.assert_methods_equal(
ORIGINAL_ROOT / "database.py",
"ReviewDatabase",
APP_ROOT / "backend" / "llm" / "repository.py",
"LLMAuditRepositoryMixin",
LLM_REPOSITORY_METHODS,
)
def test_mentor_data_profiles_are_exact_original_values(self) -> None:
names = {"MENTOR_DATA_PROFILES", "MENTOR_INDEX_UNIVERSE", "MENTOR_ETF_UNIVERSE"}
self.assertEqual(
assignments(ORIGINAL_ROOT / "server.py", names),
assignments(
APP_ROOT / "backend" / "features" / "mentor" / "service.py",
names,
),
)
def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
remaining_service = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
remaining_http = class_methods(
APP_ROOT / "backend" / "application.py", "RequestHandler"
)
self.assertTrue(MENTOR_SERVICE_METHODS.isdisjoint(remaining_service))
self.assertTrue(LLM_SERVICE_METHODS.isdisjoint(remaining_service))
self.assertTrue(MENTOR_REPOSITORY_METHODS.isdisjoint(remaining_database))
self.assertTrue(LLM_REPOSITORY_METHODS.isdisjoint(remaining_database))
self.assertTrue(
{"stream_mentor_chat", "save_llm_settings", "save_llm_mode", "test_llm_settings"}
.isdisjoint(remaining_http)
)
def test_http_mixins_preserve_stream_and_model_endpoints(self) -> None:
mentor_http = class_methods(
APP_ROOT / "backend" / "features" / "mentor" / "http.py",
"MentorHttpMixin",
)
llm_http = class_methods(APP_ROOT / "backend" / "llm" / "http.py", "LLMHttpMixin")
self.assertEqual(set(mentor_http), {"stream_mentor_chat"})
self.assertEqual(
set(llm_http), {"save_llm_settings", "save_llm_mode", "test_llm_settings"}
)
if __name__ == "__main__":
unittest.main()
@@ -1,212 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import assistant_agent
from backend.features.review import agent as canonical_agent
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
ALERT_SERVICE_METHODS = {
"alert_center",
"create_alert",
"mark_alert_read",
"mark_all_alerts_read",
"delete_alert",
}
REVIEW_SERVICE_METHODS = {
"trade_entries",
"review_watchlist",
"save_trade_entry",
"delete_trade_entry",
"assistant_messages",
"clear_assistant_messages",
"assistant_stream",
"_assistant_context",
}
ALERT_REPOSITORY_METHODS = {
"save_alert",
"list_alerts",
"count_unread_alerts",
"mark_alert_read",
"mark_all_alerts_read",
"delete_alert",
}
REVIEW_REPOSITORY_METHODS = {
"list_watchlist",
"save_watchlist",
"watchlist_price_history",
"delete_watchlist",
"list_notes",
"save_note",
"delete_note",
"save_trade_entry",
"list_trade_entries",
"delete_trade_entry",
"save_assistant_exchange",
"list_assistant_messages",
"delete_assistant_messages",
}
ALERT_HTTP_METHODS = {"save_alert"}
REVIEW_HTTP_METHODS = {
"save_trade_entry",
"stream_assistant_chat",
"save_watchlist",
"save_note",
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def top_level_definition(path: Path, name: str) -> str:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
node = next(
item
for item in tree.body
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
and item.name == name
)
return ast.dump(node, include_attributes=False)
class ReviewAlertsSliceSourceEquivalenceTests(unittest.TestCase):
def assert_methods_equal(
self,
original_path: Path,
original_class: str,
migrated_path: Path,
migrated_class: str,
expected: set[str],
) -> None:
original = class_methods(original_path, original_class)
migrated = class_methods(migrated_path, migrated_class)
self.assertEqual(set(migrated), expected)
for name in sorted(expected):
self.assertEqual(migrated[name], original[name], name)
def test_review_assistant_agent_uses_shared_transport(self) -> None:
source = (
APP_ROOT / "backend" / "features" / "review" / "agent.py"
).read_text(encoding="utf-8")
self.assertIn("llm_transport.stream_chat_completion", source)
self.assertNotIn("urllib.request", source)
def test_review_assistant_compatibility_module_is_canonical(self) -> None:
self.assertIs(assistant_agent, canonical_agent)
def test_alert_and_review_service_methods_are_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "server.py",
"DashboardService",
APP_ROOT / "backend" / "features" / "alerts" / "facade.py",
"AlertServiceMixin",
ALERT_SERVICE_METHODS,
)
self.assert_methods_equal(
ORIGINAL_ROOT / "server.py",
"DashboardService",
APP_ROOT / "backend" / "features" / "review" / "service.py",
"ReviewServiceMixin",
REVIEW_SERVICE_METHODS,
)
def test_alert_and_review_repository_methods_are_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "database.py",
"ReviewDatabase",
APP_ROOT / "backend" / "features" / "alerts" / "repository.py",
"AlertRepositoryMixin",
ALERT_REPOSITORY_METHODS,
)
self.assert_methods_equal(
ORIGINAL_ROOT / "database.py",
"ReviewDatabase",
APP_ROOT / "backend" / "features" / "review" / "repository.py",
"ReviewRepositoryMixin",
REVIEW_REPOSITORY_METHODS,
)
def test_existing_alert_and_trade_journal_services_preserve_original_classes(self) -> None:
self.assertEqual(
top_level_definition(
ORIGINAL_ROOT / "backend" / "features" / "alerts" / "service.py",
"AlertService",
),
top_level_definition(
APP_ROOT / "backend" / "features" / "alerts" / "service.py",
"AlertService",
),
)
self.assertEqual(
top_level_definition(
ORIGINAL_ROOT / "backend" / "features" / "review" / "trade_journal.py",
"TradeJournalService",
),
top_level_definition(
APP_ROOT / "backend" / "features" / "review" / "trade_journal.py",
"TradeJournalService",
),
)
def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
remaining_service = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
remaining_http = class_methods(
APP_ROOT / "backend" / "application.py", "RequestHandler"
)
self.assertTrue(ALERT_SERVICE_METHODS.isdisjoint(remaining_service))
self.assertTrue(REVIEW_SERVICE_METHODS.isdisjoint(remaining_service))
self.assertTrue(ALERT_REPOSITORY_METHODS.isdisjoint(remaining_database))
self.assertTrue(REVIEW_REPOSITORY_METHODS.isdisjoint(remaining_database))
self.assertTrue(ALERT_HTTP_METHODS.isdisjoint(remaining_http))
self.assertTrue(REVIEW_HTTP_METHODS.isdisjoint(remaining_http))
def test_http_mixins_preserve_all_endpoints_without_global_service(self) -> None:
alerts = class_methods(
APP_ROOT / "backend" / "features" / "alerts" / "http.py",
"AlertHttpMixin",
)
review = class_methods(
APP_ROOT / "backend" / "features" / "review" / "http.py",
"ReviewHttpMixin",
)
self.assertEqual(set(alerts), ALERT_HTTP_METHODS)
self.assertEqual(set(review), REVIEW_HTTP_METHODS)
for path in (
APP_ROOT / "backend" / "features" / "alerts" / "http.py",
APP_ROOT / "backend" / "features" / "review" / "http.py",
):
source = path.read_text(encoding="utf-8")
self.assertNotIn("SERVICE.", source)
self.assertIn("self.application_service.", source)
if __name__ == "__main__":
unittest.main()
-214
View File
@@ -1,214 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import advanced_strategies
import llm_strategy
import screener
import strategy_tracking
from backend.bootstrap import config as bootstrap_config
from backend.data.numbers import finite_number
from backend.features.screener import compiler, engine, strategies, tracking
from backend.features.screener import service as screener_service
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
function_contract,
module_contract,
)
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
SCREENER_SERVICE_METHODS = {
"_strategy_missing_data",
"screener_setup",
"screener_tracking",
"add_screener_tracking",
"remove_screener_tracking",
"refresh_screener_tracking",
"sync_screener_data",
"_schedule_automatic_screeners",
"run_automatic_screeners",
"compile_screener_strategy",
"save_screener_strategy",
"delete_screener_strategy",
"run_screener",
}
SCREENER_REPOSITORY_METHODS = {
"upsert_benchmark_bars",
"upsert_daily_indicators",
"upsert_fundamental_indicators",
"upsert_moneyflow",
"upsert_earnings_events",
"daily_indicator_dates",
"fundamental_periods",
"factor_dates",
"factor_health_summary",
"load_factor_data",
"snapshot_summaries",
"save_screener_strategy",
"list_screener_strategies",
"delete_screener_strategy",
"save_screener_run",
"_screener_run_payload",
"latest_screener_run",
"latest_screener_runs",
"latest_screener_context_runs",
"screener_runs_for_date",
"get_screener_run",
"save_strategy_tracks",
"list_strategy_tracks",
"delete_strategy_track",
"load_tracking_bars",
}
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def top_level_definition(path: Path, name: str) -> str:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
node = next(
item
for item in tree.body
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
and item.name == name
)
return ast.dump(node, include_attributes=False)
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
class ScreenerSliceSourceEquivalenceTests(unittest.TestCase):
def assert_methods_equal(
self,
original_path: Path,
original_class: str,
migrated_path: Path,
migrated_class: str,
names: set[str],
) -> None:
original = class_methods(original_path, original_class)
migrated = class_methods(migrated_path, migrated_class)
self.assertEqual(set(migrated), names)
for name in sorted(names):
self.assertEqual(migrated[name], original[name], name)
def test_screener_service_is_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "server.py",
"DashboardService",
APP_ROOT / "backend" / "features" / "screener" / "service.py",
"ScreenerServiceMixin",
SCREENER_SERVICE_METHODS,
)
self.assertEqual(
top_level_definition(ORIGINAL_ROOT / "server.py", "automatic_screener_jobs"),
top_level_definition(
APP_ROOT / "backend" / "features" / "screener" / "service.py",
"automatic_screener_jobs",
),
)
self.assertEqual(screener_service.SCREENER_LIBRARY_VERSION, 8)
def test_screener_repository_is_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "database.py",
"ReviewDatabase",
APP_ROOT / "backend" / "features" / "screener" / "repository.py",
"ScreenerRepositoryMixin",
SCREENER_REPOSITORY_METHODS,
)
def test_moved_methods_are_not_duplicated(self) -> None:
service_methods = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
repository_methods = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
self.assertTrue(SCREENER_SERVICE_METHODS.isdisjoint(service_methods))
self.assertTrue(SCREENER_REPOSITORY_METHODS.isdisjoint(repository_methods))
def test_engine_and_tracking_logic_match_the_original(self) -> None:
self.assertEqual(
module_contract(
ORIGINAL_ROOT / "screener.py",
excluded_definitions={"_display_date", "_number"},
exclude_imports=True,
),
module_contract(
APP_ROOT / "backend" / "features" / "screener" / "engine.py",
excluded_definitions={"_display_date", "_number"},
exclude_imports=True,
),
)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "screener.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "finite_number"),
)
self.assertIs(engine._number, finite_number)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "screener.py", "_display_date"),
function_contract(
APP_ROOT / "backend/bootstrap/config.py", "display_compact_date"
),
)
self.assertIs(engine._display_date, bootstrap_config.display_compact_date)
self.assertEqual(
class_methods(
ORIGINAL_ROOT / "backend" / "features" / "screener" / "tracking.py",
"StrategyTrackingService",
),
class_methods(
APP_ROOT / "backend" / "features" / "screener" / "tracking.py",
"StrategyTrackingService",
),
)
def test_library_is_exact_and_compiler_uses_shared_transport(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "advanced_strategies.py"),
sha256(APP_ROOT / "backend/features/screener/strategies.py"),
)
compiler_source = (
APP_ROOT / "backend/features/screener/compiler.py"
).read_text(encoding="utf-8")
self.assertEqual(compiler_source.count("llm_transport.chat_completion"), 2)
self.assertNotIn("urllib.request", compiler_source)
def test_compatibility_modules_export_the_canonical_objects(self) -> None:
self.assertIs(screener, engine)
self.assertIs(advanced_strategies, strategies)
self.assertIs(llm_strategy, compiler)
self.assertIs(
strategy_tracking.StrategyTrackingService,
tracking.StrategyTrackingService,
)
def test_screener_frontend_assets_are_unchanged(self) -> None:
assert_frontend_runtime_matches_audited_baseline(self)
assert_moved_asset_matches(self, "styles.css", "styles/styles.css")
assert_page_prefix_matches(self, "pages/screener/page.js")
if __name__ == "__main__":
unittest.main()
@@ -1,131 +0,0 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import sentiment_engine
from backend.data.numbers import non_nan_number
from backend.features.sentiment import engine as canonical_engine
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
function_contract,
module_contract,
)
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
SENTIMENT_METHODS = {
"_enrich_dashboard_sentiment",
"sentiment_history",
}
POOL_METHODS = {
"save_reason",
"_apply_reason_overrides",
"_schedule_ifind_event_enrichment",
"_refresh_ifind_event_enrichment",
"_ifind_field",
"_ifind_row_code",
"_normalize_ifind_event_time",
"_merge_ifind_event_enrichment",
}
POOL_REPOSITORY_METHODS = {
"save_reason_override",
"reason_overrides",
}
def class_methods(path: Path, class_name: str) -> dict[str, 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: ast.dump(node, include_attributes=False)
for node in owner.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
class SentimentPoolSliceSourceEquivalenceTests(unittest.TestCase):
def test_sentiment_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "sentiment" / "service.py",
"SentimentServiceMixin",
)
self.assertEqual(set(migrated), SENTIMENT_METHODS)
for name in sorted(SENTIMENT_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_pool_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "pools" / "service.py",
"PoolServiceMixin",
)
self.assertEqual(set(migrated), POOL_METHODS)
for name in sorted(POOL_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_pool_repository_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "database.py", "ReviewDatabase")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "pools" / "repository.py",
"PoolRepositoryMixin",
)
self.assertEqual(set(migrated), POOL_REPOSITORY_METHODS)
for name in sorted(POOL_REPOSITORY_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
remaining_service = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
self.assertTrue((SENTIMENT_METHODS | POOL_METHODS).isdisjoint(remaining_service))
self.assertTrue(POOL_REPOSITORY_METHODS.isdisjoint(remaining_database))
def test_sentiment_engine_is_exact_original_with_legacy_alias(self) -> None:
self.assertEqual(
module_contract(
ORIGINAL_ROOT / "sentiment_engine.py",
excluded_definitions={"_number"},
),
module_contract(
APP_ROOT / "backend" / "features" / "sentiment" / "engine.py",
excluded_definitions={"_number"},
excluded_import_modules={"backend.data.numbers"},
),
)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "sentiment_engine.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "non_nan_number"),
)
self.assertIs(sentiment_engine, canonical_engine)
self.assertIs(canonical_engine._number, non_nan_number)
def test_api_and_frontend_assets_are_unchanged(self) -> None:
self.assertEqual(
sha256(APP_ROOT / "config/api.config.json"),
sha256(ORIGINAL_ROOT / "config/api.config.json"),
)
assert_frontend_runtime_matches_audited_baseline(self)
assert_moved_asset_matches(self, "styles.css", "styles/styles.css")
for page in ("pages/sentiment/page.js", "pages/pools/page.js"):
assert_page_prefix_matches(self, page)
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from tushare_client import TushareClient
from backend.data.providers.tushare_client import TushareClient
class FakeRealtimeClient(TushareClient):
+1 -1
View File
@@ -4,7 +4,7 @@ import tempfile
import unittest
from pathlib import Path
from alert_service import AlertService
from backend.features.alerts.service import AlertService
from backend.bootstrap.container import build_application_container
from backend.database.repositories import (
SQLiteAlertRepository,
+1 -1
View File
@@ -6,7 +6,7 @@ import unittest
from pathlib import Path
from unittest.mock import patch
from assistant_agent import ReviewAssistantError, stream_review_assistant
from backend.features.review.agent import ReviewAssistantError, stream_review_assistant
from database import ReviewDatabase
from server import RequestHandler
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from sentiment_engine import _adaptive_score, _confirmed_phase
from backend.features.sentiment.engine import _adaptive_score, _confirmed_phase
class SentimentEngineTests(unittest.TestCase):
+1 -1
View File
@@ -5,7 +5,7 @@ import unittest
from pathlib import Path
from database import ReviewDatabase
from strategy_tracking import StrategyTrackingService
from backend.features.screener.tracking import StrategyTrackingService
class StrategyTrackingTests(unittest.TestCase):
+1 -1
View File
@@ -5,7 +5,7 @@ import unittest
from pathlib import Path
from database import ReviewDatabase
from trade_journal import TradeJournalService
from backend.features.review.trade_journal import TradeJournalService
class TradeJournalTests(unittest.TestCase):