xiaobaifupan/app/tests/test_preservation_slice_review_alerts.py

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_is_an_exact_file(self) -> None:
        self.assertEqual(
            sha256(ORIGINAL_ROOT / "assistant_agent.py"),
            sha256(APP_ROOT / "backend" / "features" / "review" / "agent.py"),
        )

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