awoooi/apps/api/tests/test_backup_notification_policy.py

from __future__ import annotations

import json

import pytest

from src.services.backup_notification_policy import load_latest_backup_notification_policy


def test_load_latest_backup_notification_policy_reads_newest_file(tmp_path):
    older = _snapshot(generated_at="2026-06-03T00:00:00+08:00", completion=99)
    newer = _snapshot(generated_at="2026-06-04T00:00:00+08:00", completion=100)
    (tmp_path / "backup_notification_policy_2026-06-03.json").write_text(
        json.dumps(older),
        encoding="utf-8",
    )
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(newer),
        encoding="utf-8",
    )

    loaded = load_latest_backup_notification_policy(tmp_path)

    assert loaded["generated_at"] == "2026-06-04T00:00:00+08:00"
    assert loaded["program_status"]["overall_completion_percent"] == 100
    assert loaded["rollups"]["total_rules"] == 3
    assert loaded["operation_boundaries"]["notification_send_allowed"] is False


def test_backup_notification_policy_requires_read_only_mode(tmp_path):
    snapshot = _snapshot()
    snapshot["program_status"]["read_only_mode"] = False
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="read_only_mode"):
        load_latest_backup_notification_policy(tmp_path)


def test_backup_notification_policy_requires_blocked_operations(tmp_path):
    snapshot = _snapshot()
    snapshot["operation_boundaries"]["notification_send_allowed"] = True
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="operation boundaries"):
        load_latest_backup_notification_policy(tmp_path)


def test_backup_notification_policy_requires_total_consistency(tmp_path):
    snapshot = _snapshot()
    snapshot["rollups"]["total_rules"] = 999
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="total_rules"):
        load_latest_backup_notification_policy(tmp_path)


def test_backup_notification_policy_requires_decision_rollup_consistency(tmp_path):
    snapshot = _snapshot()
    snapshot["rollups"]["by_decision"] = {"suppress_immediate_success": 3}
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="by_decision"):
        load_latest_backup_notification_policy(tmp_path)


def test_backup_notification_policy_requires_success_suppression(tmp_path):
    snapshot = _snapshot()
    snapshot["policy_rules"][0]["decision"] = "escalate_immediate"
    snapshot["rollups"]["by_decision"] = {
        "escalate_immediate": 2,
        "create_action_required": 1,
    }
    snapshot["rollups"]["immediate_escalation_rule_ids"] = [
        "scheduled_backup_success",
        "backup_failed",
    ]
    snapshot["rollups"]["suppressed_success_rule_ids"] = []
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="success rules"):
        load_latest_backup_notification_policy(tmp_path)


def test_backup_notification_policy_requires_summary_success_suppression(tmp_path):
    snapshot = _snapshot()
    snapshot["daily_summary_contract"]["success_immediate_notifications_allowed"] = True
    (tmp_path / "backup_notification_policy_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="daily summary"):
        load_latest_backup_notification_policy(tmp_path)


def test_backup_notification_policy_fails_when_missing(tmp_path):
    with pytest.raises(FileNotFoundError):
        load_latest_backup_notification_policy(tmp_path)


def _snapshot(
    *,
    generated_at: str = "2026-06-04T00:00:00+08:00",
    completion: int = 100,
) -> dict:
    return {
        "schema_version": "backup_notification_policy_v1",
        "generated_at": generated_at,
        "source_readiness_matrix_ref": "docs/evaluations/backup_dr_readiness_matrix_2026-06-04.json",
        "source_refs": ["docs/runbooks/BACKUP-STATUS.md"],
        "program_status": {
            "overall_completion_percent": completion,
            "current_priority": "P1",
            "current_task_id": "P1-103",
            "next_task_id": "P1-104",
            "read_only_mode": True,
        },
        "rollups": {
            "total_rules": 3,
            "by_decision": {
                "suppress_immediate_success": 1,
                "escalate_immediate": 1,
                "create_action_required": 1,
            },
            "immediate_escalation_rule_ids": ["backup_failed"],
            "suppressed_success_rule_ids": ["scheduled_backup_success"],
        },
        "notification_channels": [
            _channel("telegram_ops", immediate_allowed=True, requires_operator_action=True),
            _channel("daily_status_summary", immediate_allowed=False, requires_operator_action=False),
        ],
        "policy_rules": [
            _rule("scheduled_backup_success", "success", "info", "suppress_immediate_success"),
            _rule("backup_failed", "failed", "critical", "escalate_immediate"),
            _rule("metric_binding_gap", "needs_metric_binding", "warning", "create_action_required"),
        ],
        "daily_summary_contract": {
            "summary_time_taipei": "06:05",
            "success_immediate_notifications_allowed": False,
            "success_signal_sources": ["Prometheus textfile"],
            "failure_rows_require_action_refs": True,
            "mandatory_sections": ["latest successful backup targets"],
        },
        "agent_roles": [
            {
                "agent_id": "openclaw",
                "role": "arbitrate",
                "allowed_actions": ["read-only arbitration"],
                "blocked_actions": ["send notification"],
            }
        ],
        "operation_boundaries": {
            "read_only_policy_allowed": True,
            "notification_send_allowed": False,
            "backup_execution_allowed": False,
            "restore_execution_allowed": False,
            "offsite_sync_execution_allowed": False,
            "credential_marker_write_allowed": False,
            "schedule_change_allowed": False,
            "workflow_write_allowed": False,
            "telegram_test_message_allowed": False,
        },
        "approval_boundaries": {
            "sdk_installation_allowed": False,
            "paid_api_call_allowed": False,
            "shadow_or_canary_allowed": False,
            "production_routing_allowed": False,
            "destructive_operation_allowed": False,
        },
    }


def _channel(channel_id: str, *, immediate_allowed: bool, requires_operator_action: bool) -> dict:
    return {
        "channel_id": channel_id,
        "purpose": "test",
        "immediate_allowed": immediate_allowed,
        "success_immediate_allowed": False,
        "requires_operator_action": requires_operator_action,
    }


def _rule(rule_id: str, state: str, severity: str, decision: str) -> dict:
    return {
        "rule_id": rule_id,
        "event_kind": rule_id,
        "backup_state": state,
        "severity": severity,
        "decision": decision,
        "channels": ["daily_status_summary"],
        "owner_agent": "hermes",
        "requires_incident": decision == "escalate_immediate",
        "requires_approval_record": decision == "create_action_required",
        "message_contract": "test",
        "evidence_refs": ["docs/runbooks/BACKUP-STATUS.md"],
    }