tests/test_trends.py

import unittest

import numpy as np

from quread.trends import (
    alerts_to_csv,
    compute_drift_alerts,
    compute_metric_trends,
    parse_calibration_snapshots_text,
)


class TrendMetricsTest(unittest.TestCase):
    def test_parse_snapshots_supports_json_list(self):
        text = '[{"timestamp":"t0","qubits":{"0":{"gate_error":0.01}}},{"timestamp":"t1","qubits":{"0":{"gate_error":0.02}}}]'
        snapshots, meta = parse_calibration_snapshots_text(text)
        self.assertEqual(len(snapshots), 2)
        self.assertEqual(meta["format"], "json:list")
        self.assertEqual(meta["parsed"], 2)
        self.assertEqual(meta["skipped"], 0)

    def test_parse_snapshots_supports_jsonl_with_skips(self):
        text = '\n'.join(
            [
                '{"timestamp":"t0","qubits":{"0":{"gate_error":0.01}}}',
                'not-json',
                '{"timestamp":"t1","qubits":{"0":{"gate_error":0.03}}}',
            ]
        )
        snapshots, meta = parse_calibration_snapshots_text(text)
        self.assertEqual(len(snapshots), 2)
        self.assertEqual(meta["format"], "jsonl")
        self.assertGreaterEqual(meta["skipped"], 1)

    def test_compute_metric_trends_returns_series_and_ranking(self):
        csv_text = "\n".join(
            [
                "step,gate,target,control,theta",
                "0,H,0,,",
                "1,CNOT,1,0,",
            ]
        )
        snapshots_text = (
            '[{"timestamp":"t0","qubits":{"0":{"gate_error":0.01,"readout_error":0.02,"t1_us":80,"t2_us":60,"fidelity":0.99},'
            '"1":{"gate_error":0.01,"readout_error":0.02,"t1_us":80,"t2_us":60,"fidelity":0.99}}},'
            '{"timestamp":"t1","qubits":{"0":{"gate_error":0.20,"readout_error":0.10,"t1_us":20,"t2_us":15,"fidelity":0.80},'
            '"1":{"gate_error":0.01,"readout_error":0.02,"t1_us":80,"t2_us":60,"fidelity":0.99}}}]'
        )

        series, labels, ranking, meta = compute_metric_trends(
            csv_text,
            n_qubits=2,
            snapshots_text=snapshots_text,
            metric="composite",
            state_vector=np.array([1.0 + 0j, 0j, 0j, 0j], dtype=complex),
        )

        self.assertEqual(series.shape, (2, 2))
        self.assertEqual(labels, ["t0", "t1"])
        self.assertEqual(meta["metric"], "composite_risk")
        self.assertEqual(int(ranking[0]["qubit"]), 0)
        self.assertGreater(float(ranking[0]["delta"]), 0.0)
        self.assertTrue(float(ranking[0]["latest_risk"]) >= float(ranking[1]["latest_risk"]))

    def test_compute_metric_trends_inverts_risk_for_fidelity(self):
        csv_text = "\n".join(
            [
                "step,gate,target,control,theta",
                "0,H,0,,",
            ]
        )
        snapshots_text = (
            '[{"timestamp":"t0","qubits":{"0":{"fidelity":0.99},"1":{"fidelity":0.99}}},'
            '{"timestamp":"t1","qubits":{"0":{"fidelity":0.70},"1":{"fidelity":0.98}}}]'
        )
        _series, _labels, ranking, meta = compute_metric_trends(
            csv_text,
            n_qubits=2,
            snapshots_text=snapshots_text,
            metric="fidelity",
            state_vector=np.array([1.0 + 0j, 0j, 0j, 0j], dtype=complex),
        )
        self.assertEqual(meta["risk_mode"], "lower_is_risk")
        self.assertEqual(int(ranking[0]["qubit"]), 0)
        self.assertGreater(float(ranking[0]["risk_delta"]), 0.0)

    def test_compute_drift_alerts_flags_warning_and_critical(self):
        ranking = [
            {
                "qubit": 0,
                "latest_risk": 0.82,
                "risk_delta": 0.2,
                "risk_slope": 0.07,
            },
            {
                "qubit": 1,
                "latest_risk": 0.5,
                "risk_delta": 0.05,
                "risk_slope": 0.01,
            },
            {
                "qubit": 2,
                "latest_risk": 0.2,
                "risk_delta": 0.01,
                "risk_slope": 0.001,
            },
        ]
        alerts = compute_drift_alerts(
            ranking,
            warning_threshold=0.45,
            critical_threshold=0.70,
            delta_warning=0.08,
            delta_critical=0.18,
            slope_warning=0.02,
            slope_critical=0.05,
        )
        self.assertEqual(alerts[0]["qubit"], 0)
        self.assertEqual(alerts[0]["level"], "critical")
        self.assertEqual(alerts[1]["level"], "warning")
        self.assertEqual(alerts[2]["level"], "ok")

    def test_alerts_to_csv_contains_rows_and_triggers(self):
        alerts = [
            {
                "qubit": 0,
                "level": "critical",
                "latest_risk": 0.82,
                "risk_delta": 0.2,
                "risk_slope": 0.07,
                "triggers": ["latest_risk>=critical", "risk_delta>=critical"],
            }
        ]
        csv_text = alerts_to_csv(alerts)
        self.assertIn("qubit,level,latest_risk,risk_delta,risk_slope,triggers", csv_text)
        self.assertIn("0,critical,0.82,0.2,0.07", csv_text)
        self.assertIn("latest_risk>=critical; risk_delta>=critical", csv_text)


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