cyb50-quant/research/dragon/v2/tests/test_rollout_governance.py

from __future__ import annotations

import unittest

import pandas as pd

from dragon_rollout_governance import (
    FORWARD_OK,
    HOLD_AND_REVIEW,
    ROLLBACK_REVIEW_REQUIRED,
    evaluate_rollout,
)


class TestRolloutGovernance(unittest.TestCase):
    def _manifest(self, latest_bar_date: str = "2026-04-08") -> dict[str, object]:
        return {
            "release_version": "RC1",
            "branch": "alpha_first_glued_refined_hot_cap",
            "as_of_request_date": latest_bar_date,
            "latest_bar_date": latest_bar_date,
        }

    def _monitor_history(self, latest_bar_date: str = "2026-04-08", status: str = "ok") -> pd.DataFrame:
        return pd.DataFrame(
            [
                {
                    "latest_bar_date": latest_bar_date,
                    "metric": "next_open_avg_return_delta_vs_control",
                    "actual_value": 0.01,
                    "status": status,
                    "warning_streak": 0 if status == "ok" else 3,
                    "hard_breach_streak": 0 if status != "hard_breach" else 1,
                }
            ]
        )

    def _divergence_log(self, latest_bar_date: str = "2026-04-08", divergence_level: str = "none") -> pd.DataFrame:
        return pd.DataFrame(
            [
                {
                    "latest_bar_date": latest_bar_date,
                    "divergence_level": divergence_level,
                    "same_position_flag": True,
                    "same_latest_real_event_flag": True,
                }
            ]
        )

    def test_forward_ok_when_all_gates_pass(self) -> None:
        decision, gates, reasons, active_branch = evaluate_rollout(
            manifest=self._manifest(),
            monitor_history=self._monitor_history(),
            divergence_log=self._divergence_log(),
            monitor_health_report_exists=True,
        )
        self.assertEqual(decision, FORWARD_OK)
        self.assertEqual(reasons, [])
        self.assertEqual(active_branch, "alpha_first_glued_refined_hot_cap")
        self.assertTrue(all(g.status == "ok" for g in gates))

    def test_hard_breach_requires_rollback_review(self) -> None:
        decision, gates, reasons, active_branch = evaluate_rollout(
            manifest=self._manifest(),
            monitor_history=self._monitor_history(status="hard_breach"),
            divergence_log=self._divergence_log(),
            monitor_health_report_exists=True,
        )
        self.assertEqual(decision, ROLLBACK_REVIEW_REQUIRED)
        self.assertTrue(any("hard_breach_budget" in reason for reason in reasons))
        self.assertEqual(active_branch, "alpha_first_selective_veto")
        self.assertTrue(any(g.status == "hard_fail" for g in gates))

    def test_material_divergence_or_warning_moves_to_hold(self) -> None:
        monitor = pd.DataFrame(
            [
                {
                    "latest_bar_date": "2026-04-08",
                    "metric": "next_open_avg_return_delta_vs_control",
                    "actual_value": 0.01,
                    "status": "warning",
                    "warning_streak": 1,
                    "hard_breach_streak": 0,
                },
                {
                    "latest_bar_date": "2026-04-08",
                    "metric": "next_open_profit_factor_delta_vs_control",
                    "actual_value": 0.20,
                    "status": "warning",
                    "warning_streak": 1,
                    "hard_breach_streak": 0,
                },
            ]
        )
        decision, gates, reasons, _ = evaluate_rollout(
            manifest=self._manifest(),
            monitor_history=monitor,
            divergence_log=self._divergence_log(divergence_level="material"),
            monitor_health_report_exists=True,
        )
        self.assertEqual(decision, HOLD_AND_REVIEW)
        self.assertTrue(any(g.status == "warning" for g in gates))
        self.assertTrue(any(reason.startswith("warning_budget") or reason.startswith("branch_divergence_level") for reason in reasons))


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