import json import re import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] def read(path: str) -> str: return (ROOT / path).read_text(encoding="utf-8") def active_route_source() -> str: lines = read("frontend/src/router/routes.js").splitlines() return "\n".join(line for line in lines if not line.lstrip().startswith("//")) class CleanupContractTest(unittest.TestCase): def test_retained_frontend_capabilities_exist(self): retained = [ "frontend/src/views/home/index.vue", "frontend/src/views/home/edit.vue", "frontend/src/views/home/workbench.js", "frontend/src/views/knowledgeBaseProduct/index.vue", "frontend/src/views/dataGovernance/dataSecurity/index.vue", "frontend/src/views/dataGovernance/dataProcess/index.vue", "frontend/src/views/dataFactory/workflow/index.vue", "frontend/src/views/dataService/dataOrder/index.vue", "frontend/src/views/dataService/dataProduct/index.vue", "frontend/src/views/dataReview/index.vue", "frontend/src/views/systemManage/error/404.vue", ] for path in retained: with self.subTest(path=path): self.assertTrue((ROOT / path).is_file(), path) routes = active_route_source() for component in [ "dataGovernance/dataSecurity", "dataGovernance/dataProcess", "dataFactory/workflow", "dataService/dataOrder", "dataService/dataProduct", "dataReview", ]: with self.subTest(component=component): self.assertIn(component, routes) def test_confirmed_legacy_frontend_directories_are_removed(self): removed = [ "frontend/src/views/dataChart", "frontend/src/views/modelSystem", "frontend/src/views/dataAsset", "frontend/src/views/dataDiagram", "frontend/src/views/questionnaireInquiry", "frontend/src/views/knowledgeBaseResume", "frontend/src/views/dataGovernance/dataResource", "frontend/src/views/dataGovernance/dataModules", "frontend/src/views/dataGovernance/dataIndicator", "frontend/src/views/dataBook/dataResource", "frontend/src/views/dataBook/dataModel", "frontend/src/views/dataBook/indicator", "frontend/src/views/dataFactory/dispatch", "frontend/src/views/dataFactory/logManage", "frontend/src/views/dataFactory/nodeManage", "frontend/src/views/dataFactory/productionLineMonitor", "frontend/src/components/ChartTemplate/humanResource", "frontend/src/components/phoneVerify", "frontend/src/components/UnifyExport", "frontend/src/views/login/components/forget.vue", "frontend/src/views/login/components/register.vue", ] for path in removed: with self.subTest(path=path): self.assertFalse((ROOT / path).exists(), path) def test_confirmed_legacy_routes_are_removed(self): source = read("frontend/src/router/routes.js") removed_tokens = [ "/data-chart", "dataAsset", "/model-system", "dataGovernance/dataResource", "dataGovernance/dataModules", "dataGovernance/dataIndicator", "dataBook/dataResource", "dataBook/dataModel", "dataBook/indicator", ] for token in removed_tokens: with self.subTest(token=token): self.assertNotIn(token, source) def test_frontend_api_surface_is_slimmed(self): for path in [ "frontend/src/api/dataChart.js", "frontend/src/api/chartCustomized.js", "frontend/src/api/report.js", "frontend/src/api/productionLine.js", "frontend/src/api/dataBook.js", "frontend/src/api/system.js", "frontend/src/api/menu.js", ]: with self.subTest(path=path): self.assertFalse((ROOT / path).exists(), path) data_factory = read("frontend/src/api/dataFactory.js") exports = set(re.findall(r"^export function (\w+)", data_factory, re.MULTILINE)) self.assertEqual( exports, { "getWorkflows", "getWorkflow", "getWorkflowStatus", "activateWorkflow", "deactivateWorkflow", "getWorkflowExecutions", "getAllExecutions", "getExecution", "triggerWorkflow", "healthCheck", }, ) governance = read("frontend/src/api/dataGovernance.js") for token in ["const dataResource", "const dataModel", "const dataIndicator"]: self.assertNotIn(token, governance) for token in ["runDataFlow", "getDataFlowExecute", "getDataFlowLog"]: self.assertNotIn(token, governance) data_origin = read("frontend/src/api/dataOrigin.js") self.assertNotIn("businessCard", data_origin) self.assertIn("getDatasourceList", data_origin) user_api = read("frontend/src/api/user.js") self.assertEqual( set(re.findall(r"^export function (\w+)", user_api, re.MULTILINE)), {"login", "getCurrentUser"}, ) workbench = read("frontend/src/views/home/workbench.js") for widget_id in [ "pending-review", "order-status", "product-statistics", "datasource-health", "n8n-execution-status", ]: self.assertIn(widget_id, workbench) def test_backend_runtime_and_obsolete_subsystems_are_removed(self): routes = read("app/api/data_flow/routes.py") service = read("app/core/data_flow/dataflows.py") for token in [ "execute-dataflow", "get-dataflow-status", "get-dataflow-logs", ]: self.assertNotIn(token, routes) for token in [ "def execute_dataflow(", "def get_dataflow_status(", "def get_dataflow_logs(", ]: self.assertNotIn(token, service) self.assertFalse((ROOT / "mcp-servers/task-manager").exists()) self.assertNotIn("AIRFLOW_", read("app/config/config.py")) self.assertNotIn("AIRFLOW_", read("deployment/dataops.env")) self.assertNotIn("/auth/register", read("app/api/system/routes.py")) removed_sql = [ "create_calendar_info.sql", "create_calendar_records.sql", "create_wechat_users.sql", "create_duplicate_business_cards_table.sql", "hotel_group_brands_ddl.sql", "hotel_positions_ddl.sql", ] for filename in removed_sql: self.assertFalse((ROOT / "database" / filename).exists(), filename) def test_n8n_and_dataflow_definition_contracts_remain(self): factory_routes = read("app/api/data_factory/routes.py") for token in ["/workflows", "/executions", "/health"]: self.assertIn(token, factory_routes) flow_routes = read("app/api/data_flow/routes.py") for token in [ "/get-dataflows-list", "/get-dataflow/", "/add-dataflow", "/update-dataflow/", "/delete-dataflow/", "/get-BD-list", "/get-script/", ]: self.assertIn(token, flow_routes) def test_hopms_acceptance_assets_remain(self): retained = [ "scripts/import_hopms_dataset.py", "tests/test_hopms_import.py", "docs/HOPMs标准数据集-V1.1.3.docx", "docs/generated/hopms_dry_run.json", "docs/generated/hopms_import_result.json", ] for path in retained: with self.subTest(path=path): self.assertTrue((ROOT / path).is_file(), path) def test_frontend_targets_supported_node_lts(self): package = json.loads(read("frontend/package.json")) engine = package["engines"]["node"] self.assertTrue(engine.startswith(">=24") or engine.startswith("24"), engine) if __name__ == "__main__": unittest.main()