""" 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}"