xiaobai-review/tests/test_preservation_slice_market_insights.py

from __future__ import annotations

import ast
import hashlib
import unittest
from pathlib import Path

import market_insights
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,
)


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)

    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()