130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
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",
|
|
"_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()
|