test_poc.py

"""Regression tests for the topology-neutral process contract and central claims."""

import json
import unittest

import pandas as pd

import analyzer
import config
from simulator import simulate_batch
from topology import product_config, vessel_config


class TopologyContractTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        with open("data/topology_contract_manifest.json", encoding="utf-8") as file:
            cls.manifest = json.load(file)
        with open("data/results.json", encoding="utf-8") as file:
            cls.results = json.load(file)
        cls.process = pd.read_csv("data/topology_process_contract.csv")

    # --- Physics and production ---
    def test_batch_generation_is_reproducible_and_physically_bounded(self):
        tank = vessel_config("Topology_A", 0)
        product = product_config("Ale")
        first = simulate_batch(tank, product, "COLD_CRASH", 11)
        second = simulate_batch(tank, product, "COLD_CRASH", 11)
        pd.testing.assert_frame_equal(first, second)
        self.assertGreaterEqual(
            first[tank["tag_temp"]].min(),
            config.GLYCOL_SUPPLY_TEMP_C - 0.5,
        )

    def test_train_and_test_splits_do_not_overlap(self):
        train_runs = set(self.process.loc[self.process["split"] == "train", "run_id"])
        test_runs = set(self.process.loc[self.process["split"] == "test", "run_id"])
        self.assertTrue(train_runs.isdisjoint(test_runs))
        self.assertFalse(
            (analyzer.training_mask(self.process) & analyzer.evaluation_mask(self.process)).any()
        )

    def test_each_topology_has_its_declared_distinct_volume(self):
        expected = {"Topology_A": 200.0, "Topology_B": 400.0, "Topology_C": 300.0}
        self.assertEqual(config.TOPOLOGY_SITE_VOLUMES_HL, expected)
        self.assertEqual(len(set(expected.values())), len(expected))
        for site_id, volume in expected.items():
            site_rows = self.process[self.process["site_id"] == site_id]
            self.assertTrue(site_rows["active_volume_hl"].eq(volume).all())
        self.assertTrue(
            self.process.groupby("run_id").size().eq(config.TOTAL_STEPS).all()
        )

    # --- Contract structure ---
    def test_one_contract_replaces_three_asset_adapters(self):
        self.assertFalse(self.manifest["raw_schemas_identical"])
        self.assertEqual(
            self.manifest["raw_feature_counts"],
            {"Topology_A": 5, "Topology_B": 17, "Topology_C": 7},
        )
        self.assertEqual(self.manifest["asset_specific_adapter_count"], 3)
        self.assertEqual(self.manifest["generic_process_aggregator_count"], 1)
        self.assertTrue(
            all(self.manifest["canonical_outputs_numerically_equal"].values())
        )

    def test_shared_utility_granularity_is_not_fabricated(self):
        topology_c = self.process[self.process["site_id"] == "Topology_C"]
        self.assertEqual(self.manifest["shared_utility_site"], "Topology_C")
        self.assertFalse(self.manifest["shared_utility_allocation_available"])
        self.assertTrue(topology_c["utility_scope"].eq("shared").all())
        self.assertFalse(topology_c["allocation_to_equipment_available"].all())

    def test_registry_omits_curated_engineering_attributes(self):
        registry = pd.read_csv("data/asset_registry_raw.csv")
        curated = pd.read_csv("data/curated_engineering_attributes.csv")
        for field in config.CURATED_ATTRIBUTE_FIELDS:
            self.assertNotIn(field, registry.columns)
            self.assertIn(field, curated.columns)
        self.assertTrue(self.manifest["curated_attributes"]["intensive_features_need_curation"])

    def test_ownership_reuse_ratio_scales_with_sites(self):
        ownership = self.manifest["ownership_cost"]
        self.assertEqual(ownership["asset_definitions_now"], len(config.TOPOLOGY_SITE_CONFIGS))
        self.assertEqual(ownership["process_definitions_now"], 1)
        self.assertEqual(ownership["definition_reuse_ratio"], float(len(config.TOPOLOGY_SITE_CONFIGS)))

    # --- Core claim: data layer ---
    def test_raw_asset_model_collapses_but_process_contract_transfers(self):
        raw_transfer = self.results["asset_centric"]["raw_transfer"]
        a_on_a = raw_transfer["a_on_a_raw"]
        a_on_b = raw_transfer["a_on_b_raw"]
        global_b = self.results["process_centric"]["global_a_b_model"]["Topology_B"]
        self.assertGreater(a_on_a["macro_f1"], 0.75)
        self.assertLess(a_on_b["macro_f1"], 0.60)
        self.assertGreater(a_on_b["fpr"], 0.90)
        self.assertGreater(global_b["macro_f1"], 0.90)
        self.assertGreater(a_on_a["macro_f1"] - a_on_b["macro_f1"], 0.3)

    def test_raw_transfer_uses_native_unit_distribution_shift(self):
        shift = self.results["asset_centric"]["raw_transfer"]["distribution_shift"]
        self.assertEqual(shift["train_temp_unit"], "C")
        self.assertEqual(shift["test_temp_unit"], "F")
        self.assertEqual(shift["train_actuator_type"], "analog")
        self.assertEqual(shift["test_actuator_type"], "solenoid")
        self.assertGreater(shift["test_temp_mean"], shift["train_temp_mean"] + 20)

    def test_multi_topology_training_improves_b_transfer(self):
        process = self.results["process_centric"]
        a_only_b = process["single_topology_a_model"]["Topology_B"]
        global_b = process["global_a_b_model"]["Topology_B"]
        self.assertGreater(global_b["macro_f1"], a_only_b["macro_f1"])
        self.assertLess(global_b["fpr"], a_only_b["fpr"])
        self.assertGreater(global_b["macro_f1"], 0.90)

    def test_global_training_excludes_unseen_topology_c(self):
        process = self.results["process_centric"]
        self.assertEqual(process["training_topologies"], ["Topology_A", "Topology_B"])
        self.assertEqual(process["unseen_topology"], "Topology_C")

    # --- Enrichments (Phase 2/3) ---
    def test_product_matrix_detects_both_products(self):
        grid = self.results["product_matrix"]["global_process_model"]
        for product_id in config.PRODUCT_IDS:
            self.assertGreater(grid["Topology_B"][product_id]["macro_f1"], 0.85)

    def test_ablation_unit_conversion_recovers_temperature_transfer(self):
        steps = self.results["ablation"]["steps"]
        self.assertGreater(steps["+unit_convert"]["macro_f1"], steps["native"]["macro_f1"] + 0.2)

    def test_calibration_reduces_unseen_site_false_positives(self):
        curve = self.results["calibration"]["curve"]
        self.assertLessEqual(curve[-1]["fpr"], curve[0]["fpr"])

    def test_blend_genealogy_is_detected_at_lot_level(self):
        genealogy = self.results["genealogy"]
        self.assertEqual(genealogy["level"], "blend_lot")
        self.assertGreater(genealogy["blend_lots"], 0)
        self.assertGreater(genealogy["metrics"]["macro_f1"], 0.8)

    def test_fault_taxonomy_covers_non_temperature_signals(self):
        taxonomy = self.results["fault_taxonomy"]
        self.assertEqual(taxonomy["STUCK_FERMENTATION"]["carrying_signal"], "gravity")
        self.assertEqual(taxonomy["SENSOR_DRIFT"]["carrying_signal"], "intra_vessel_spread")
        for fault in ("STUCK_FERMENTATION", "SENSOR_DRIFT"):
            self.assertGreater(taxonomy[fault]["recall"], 0.6)

    def test_data_layer_effect_holds_for_every_algorithm(self):
        delta = self.results["robustness"]["data_layer_delta_macro_f1"]
        self.assertEqual(set(delta), set(config.MODEL_REGISTRY))
        for algorithm, gain in delta.items():
            self.assertGreater(gain, 0.0, f"{algorithm} raw->process gain not positive")

    # --- Multi-equipment type: ProcessRun covers two equipment types ---
    def test_process_run_spans_two_equipment_types(self):
        self.assertEqual(self.results["feature_contract"]["analysis_unit"], "ProcessRun")
        self.assertEqual(
            self.results["feature_contract"]["spans_equipment_types"],
            [config.FERMENTER_ROLE, config.CONDITIONING_ROLE],
        )
        contract = pd.read_csv("data/conditioning_process_contract.csv")
        volumes = contract.groupby("site_id")["active_volume_hl"].first().to_dict()
        self.assertEqual(volumes, {"Topology_A": 200.0, "Topology_B": 400.0, "Topology_C": 300.0})
        counts = contract.groupby("site_id")["active_equipment_count"].first().to_dict()
        self.assertEqual(counts["Topology_B"], 2)

    def test_equipment_type_raw_schemas_are_disjoint(self):
        cross = self.manifest["cross_equipment"]
        self.assertTrue(cross["raw_schemas_disjoint"])
        self.assertTrue(
            set(cross["fermenter_measurement_columns"]).isdisjoint(
                cross["conditioning_measurement_columns"]
            )
        )
        self.assertEqual(cross["equipment_types"], [config.FERMENTER_ROLE, config.CONDITIONING_ROLE])

    def test_process_run_detects_conditioning_only_faults(self):
        cross = self.results["cross_equipment"]
        recall = cross["conditioning_only_fault_recall"]
        for fault in ("DIACETYL_STALL", "PRESSURE_LOSS"):
            self.assertGreater(recall[fault], 0.6, f"{fault} not detected by process run")
        by_site = cross["conditioning_process_model"]["by_site"]
        for site_id in ("Topology_A", "Topology_B", "Topology_C"):
            self.assertGreater(by_site[site_id]["macro_f1"], 0.8)

    def test_fermenter_asset_model_is_structurally_blind_to_conditioning_faults(self):
        cross = self.results["cross_equipment"]
        self.assertTrue(cross["fermenter_structurally_blind"])
        self.assertFalse(any(cross["fermenter_schema_can_detect_fault"].values()))

    def test_ownership_cost_scales_with_equipment_types(self):
        ownership = self.manifest["ownership_cost"]
        n_types = len(config.EQUIPMENT_TYPES)
        n_sites = len(config.TOPOLOGY_SITE_CONFIGS)
        self.assertEqual(ownership["asset_definitions_multi_type_now"], n_sites * n_types)
        self.assertEqual(ownership["process_definitions_multi_type_now"], 1)
        self.assertEqual(ownership["multi_type_reuse_ratio"], float(n_sites * n_types))

    def test_results_expose_all_evidence_sections(self):
        expected = {
            "generated_at",
            "contract",
            "feature_contract",
            "model_contract",
            "asset_centric",
            "process_centric",
            "product_matrix",
            "ablation",
            "calibration",
            "genealogy",
            "fault_taxonomy",
            "robustness",
            "cross_equipment",
            "ownership_cost",
            "series",
        }
        self.assertEqual(set(self.results), expected)
        self.assertEqual(self.results["feature_contract"]["analysis_unit"], "ProcessRun")


if __name__ == "__main__":
    unittest.main()