microdao-daarion/tests/test_release_gate_policy.py

"""
Tests for Release Gate Policy (GatePolicy loader + strict/off/warn behaviors).

Covers:
  1. test_gate_policy_warn_default       — no gate_profile → privacy/cost are warn, pass=True
  2. test_gate_policy_strict_privacy_fails — staging/prod + error findings → release fails
  3. test_gate_policy_off_skips           — mode=off → privacy_watch gate not in output
  4. test_gate_policy_warn_with_findings  — warn + findings → pass=True but recommendations added
  5. test_gate_policy_profile_staging     — staging profile loaded correctly
  6. test_gate_policy_profile_prod        — prod profile loaded correctly
  7. test_gate_policy_missing_file        — missing yml → graceful fallback (warn)
  8. test_strict_no_block_on_warning      — strict but fail_on=error only → warning finding ≠ block
"""

from __future__ import annotations

import asyncio
import os
import sys
import tempfile
from pathlib import Path
from typing import Dict
from unittest.mock import AsyncMock, MagicMock

import pytest

# ─── Path setup ──────────────────────────────────────────────────────────────
ROUTER_DIR = Path(__file__).parent.parent / "services" / "router"
REPO_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(ROUTER_DIR))
sys.path.insert(0, str(REPO_ROOT))

os.environ.setdefault("REPO_ROOT", str(REPO_ROOT))
os.environ["AUDIT_BACKEND"] = "memory"


# ─── Helpers ──────────────────────────────────────────────────────────────────

def _fake_tool_results(privacy_findings=None, privacy_errors=0, cost_anomalies=0):
    """Build a fake execute_tool that returns configurable gate data."""
    class FR:
        def __init__(self, data, success=True, error=None):
            self.success = success; self.result = data; self.error = error

    async def _exec(tool_name, args, agent_id=None):
        if tool_name == "pr_reviewer_tool":
            return FR({"approved": True, "verdict": "LGTM", "issues": []})
        if tool_name == "config_linter_tool":
            return FR({"pass": True, "errors": [], "warnings": []})
        if tool_name == "dependency_scanner_tool":
            return FR({"pass": True, "summary": "ok", "vulnerabilities": []})
        if tool_name == "contract_tool":
            return FR({"pass": True, "breaking_changes": [], "warnings": []})
        if tool_name == "threatmodel_tool":
            return FR({"risk_level": "low", "threats": []})
        if tool_name == "data_governance_tool":
            action = args.get("action", "")
            if action == "scan_repo":
                findings = privacy_findings or []
                e = sum(1 for f in findings if f.get("severity") == "error")
                w = sum(1 for f in findings if f.get("severity") == "warning")
                return FR({
                    "pass": True, "summary": f"{e}e {w}w",
                    "stats": {"errors": e, "warnings": w, "infos": 0},
                    "findings": findings,
                    "recommendations": (
                        ["Fix privacy errors"] if e > 0 else
                        (["Review warnings"] if w > 0 else [])
                    ),
                })
            return FR({"pass": True, "findings": [], "recommendations": [], "stats": {}})
        if tool_name == "cost_analyzer_tool":
            return FR({
                "anomalies": [{"tool": "comfy", "type": "cost_spike", "ratio": 4.0,
                                "window_calls": 60, "baseline_calls": 2,
                                "recommendation": "rate limit comfy"}] * cost_anomalies,
                "anomaly_count": cost_anomalies,
            })
        return FR({})

    return _exec


async def _run(inputs: Dict, privacy_findings=None, cost_anomalies=0):
    from release_check_runner import run_release_check, _reload_gate_policy
    _reload_gate_policy()

    tm = MagicMock()
    tm.execute_tool = AsyncMock(side_effect=_fake_tool_results(
        privacy_findings=privacy_findings,
        cost_anomalies=cost_anomalies,
    ))
    return await run_release_check(tm, inputs, agent_id="sofiia")


# ─── 1. Default (dev) — warn → pass ──────────────────────────────────────────

def test_gate_policy_warn_default():
    """No gate_profile → dev profile → warn mode → privacy/cost don't block."""
    privacy_findings = [
        {"id": "DG-LOG-001", "severity": "error", "title": "Secret logged",
         "category": "logging", "evidence": {}, "recommended_fix": ""},
    ]
    report = asyncio.run(_run(
        {"diff_text": "x", "run_privacy_watch": True, "run_cost_watch": True,
         "fail_fast": False},
        privacy_findings=privacy_findings,
    ))

    assert report["pass"] is True, "dev/warn mode: error findings should NOT block release"
    gate_names = [g["name"] for g in report["gates"]]
    assert "privacy_watch" in gate_names

    pw = next(g for g in report["gates"] if g["name"] == "privacy_watch")
    assert pw["status"] == "pass"
    # Recommendation should be in the report
    assert any("privacy" in r.lower() or "error" in r.lower() or "fix" in r.lower()
               for r in report.get("recommendations", []))


# ─── 2. Staging strict — error findings → release fails ───────────────────────

def test_gate_policy_strict_privacy_fails():
    """staging profile + strict privacy + error finding → release_check fails."""
    privacy_findings = [
        {"id": "DG-SEC-001", "severity": "error", "title": "Private key in repo",
         "category": "secrets", "evidence": {"path": "config.py", "details": "***"},
         "recommended_fix": "Remove key"},
    ]
    report = asyncio.run(_run(
        {"diff_text": "x", "gate_profile": "staging",
         "run_privacy_watch": True, "run_cost_watch": True, "fail_fast": False},
        privacy_findings=privacy_findings,
    ))

    assert report["pass"] is False, "staging strict mode: error finding must block release"
    pw = next(g for g in report["gates"] if g["name"] == "privacy_watch")
    assert pw["status"] == "pass"  # gate itself says pass (it found findings)
    # But overall_pass was set to False by strict logic


# ─── 3. gate_mode=off → privacy_watch skipped ────────────────────────────────

def test_gate_policy_off_skips():
    """privacy_watch mode=off → gate not run at all."""
    # Temporarily write a custom policy that sets privacy_watch off
    from release_check_runner import _reload_gate_policy, load_gate_policy
    import yaml

    custom_policy = {
        "profiles": {
            "custom_off": {
                "gates": {
                    "privacy_watch": {"mode": "off"},
                    "cost_watch": {"mode": "off"},
                }
            }
        },
        "defaults": {"mode": "warn"},
    }

    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".yml", delete=False, dir=str(REPO_ROOT / "config")
    ) as tmp_f:
        yaml.dump(custom_policy, tmp_f)
        tmp_name = tmp_f.name

    # Monkey-patch _GATE_POLICY_PATH
    import release_check_runner as rcr
    original_path = rcr._GATE_POLICY_PATH
    original_cache = rcr._gate_policy_cache
    rcr._GATE_POLICY_PATH = tmp_name
    rcr._gate_policy_cache = None

    try:
        report = asyncio.run(_run(
            {"diff_text": "x", "gate_profile": "custom_off",
             "run_privacy_watch": True, "run_cost_watch": True},
        ))
        gate_names = [g["name"] for g in report["gates"]]
        assert "privacy_watch" not in gate_names, "mode=off must skip the gate"
        assert "cost_watch" not in gate_names
    finally:
        rcr._GATE_POLICY_PATH = original_path
        rcr._gate_policy_cache = original_cache
        Path(tmp_name).unlink(missing_ok=True)


# ─── 4. warn mode with findings → pass=True, recommendations added ───────────

def test_gate_policy_warn_with_findings():
    """Warnings in dev profile → pass=True, recommendations in report."""
    privacy_findings = [
        {"id": "DG-LOG-001", "severity": "warning", "title": "Sensitive field logged",
         "category": "logging", "evidence": {}, "recommended_fix": "Apply redact()"},
    ]
    report = asyncio.run(_run(
        {"diff_text": "x", "gate_profile": "dev",
         "run_privacy_watch": True, "run_cost_watch": False},
        privacy_findings=privacy_findings,
    ))

    assert report["pass"] is True
    assert len(report.get("recommendations", [])) >= 1


# ─── 5. Staging profile loaded correctly ─────────────────────────────────────

def test_gate_policy_profile_staging():
    from release_check_runner import load_gate_policy, _reload_gate_policy
    _reload_gate_policy()
    policy = load_gate_policy("staging")
    pw = policy.get("privacy_watch") or {}
    assert pw.get("mode") == "strict"
    assert "error" in (pw.get("fail_on") or [])


# ─── 6. Prod profile loaded correctly ────────────────────────────────────────

def test_gate_policy_profile_prod():
    from release_check_runner import load_gate_policy, _reload_gate_policy
    _reload_gate_policy()
    policy = load_gate_policy("prod")
    pw = policy.get("privacy_watch") or {}
    assert pw.get("mode") == "strict"
    cw = policy.get("cost_watch") or {}
    assert cw.get("mode") == "warn"  # cost always warn even in prod


# ─── 7. Missing policy file → graceful fallback ───────────────────────────────

def test_gate_policy_missing_file():
    import release_check_runner as rcr
    original_path = rcr._GATE_POLICY_PATH
    original_cache = rcr._gate_policy_cache
    rcr._GATE_POLICY_PATH = "/nonexistent/path/policy.yml"
    rcr._gate_policy_cache = None

    try:
        policy = rcr.load_gate_policy("prod")
        # Should not crash; default_mode should be "warn"
        assert policy.get("_default_mode") == "warn"
    finally:
        rcr._GATE_POLICY_PATH = original_path
        rcr._gate_policy_cache = original_cache


# ─── 8. strict + fail_on=error only → warning doesn't block ──────────────────

def test_strict_no_block_on_warning_only():
    """staging strict mode + fail_on=error only → warning-level finding does NOT block."""
    privacy_findings = [
        {"id": "DG-LOG-001", "severity": "warning", "title": "Warn finding",
         "category": "logging", "evidence": {}, "recommended_fix": ""},
    ]
    report = asyncio.run(_run(
        {"diff_text": "x", "gate_profile": "staging",
         "run_privacy_watch": True, "fail_fast": False},
        privacy_findings=privacy_findings,
    ))

    # staging fail_on=["error"] only — warning should not block
    assert report["pass"] is True


# ─── 9. cost_watch always pass regardless of profile ─────────────────────────

def test_cost_watch_always_pass_all_profiles():
    """cost_watch is always warn in all profiles — never blocks release."""
    for profile in ["dev", "staging", "prod"]:
        report = asyncio.run(_run(
            {"diff_text": "x", "gate_profile": profile,
             "run_privacy_watch": False, "run_cost_watch": True},
            cost_anomalies=5,
        ))
        assert report["pass"] is True, f"cost_watch must not block in profile={profile}"