DataOps-platform/tests/test_cleanup_contract.py

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",
                "listIdentityProviders",
                "beginEnterpriseLogin",
                "exchangeIdentityCode",
                "refreshIdentitySession",
                "logoutIdentitySession",
            },
        )

        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/<int:dataflow_id>",
            "/add-dataflow",
            "/update-dataflow/<int:dataflow_id>",
            "/delete-dataflow/<int:dataflow_id>",
            "/get-BD-list",
            "/get-script/<int:dataflow_id>",
        ]:
            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()