xiaobai-review/tests/test_preservation_slice_market.py

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.data import realtime
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 (
    ORIGINAL_STATIC,
    assert_moved_asset_matches,
    reassembled_frontend_runtime,
)


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))
        self.assertEqual(
            top_level_definitions(ORIGINAL_ROOT / "tushare_client.py"),
            top_level_definitions(APP_ROOT / "backend/data/providers/tushare_client.py"),
        )
        self.assertEqual(
            top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py"),
            top_level_definitions(APP_ROOT / "backend/features/market/charts.py"),
        )

    def test_relocated_frontend_preserves_original_market_runtime_and_styles(self) -> None:
        self.assertEqual(
            reassembled_frontend_runtime(),
            (ORIGINAL_STATIC / "app.js").read_text(encoding="utf-8"),
        )
        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()