Jon Chery
274 lines
11 KiB
Python
274 lines
11 KiB
Python
"""Tests for Nova metrics event emitters (P1, REQ-187/188).
|
|
|
|
Tests the CloudEvents envelope, per-run manifest writer, Decision Ledger
|
|
(hash-chain integrity + verify-chain), and the event emission from
|
|
confidence_signal, hitl_gates, and checkov_adapter.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_metrics(tmp_path, monkeypatch):
|
|
"""Redirect metrics/ to a tmp dir for isolated testing."""
|
|
metrics_dir = tmp_path / "metrics"
|
|
metrics_dir.mkdir()
|
|
runs_dir = metrics_dir / "runs"
|
|
runs_dir.mkdir()
|
|
events_log = metrics_dir / "events.jsonl"
|
|
ledger_db = metrics_dir / "decision_ledger.db"
|
|
|
|
monkeypatch.setattr("core.metrics.event_envelope.METRICS_DIR", str(metrics_dir))
|
|
monkeypatch.setattr("core.metrics.event_envelope.EVENTS_LOG", str(events_log))
|
|
monkeypatch.setattr("core.metrics.run_manifest._METRICS_DIR", str(metrics_dir))
|
|
monkeypatch.setattr("core.metrics.run_manifest._RUNS_DIR", str(runs_dir))
|
|
monkeypatch.setattr("core.metrics.decision_ledger._LEDGER_PATH", str(ledger_db))
|
|
return {"metrics_dir": metrics_dir, "events_log": events_log, "ledger_db": ledger_db, "runs_dir": runs_dir}
|
|
|
|
|
|
# --- Task 1: CloudEvents envelope ---
|
|
|
|
def test_envelope_valid(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
ev = make_event("nova.run.completed", "run-test-1", "dev", {"exit_code": 0}, contract_id="cid-123")
|
|
assert ev["specversion"] == "1.0"
|
|
assert ev["type"] == "nova.run.completed"
|
|
assert ev["platform"]["run_id"] == "run-test-1"
|
|
assert ev["platform"]["environment"] == "dev"
|
|
assert ev["platform"]["contract_id"] == "cid-123"
|
|
assert ev["data"]["exit_code"] == 0
|
|
assert ev["datacontenttype"] == "application/json"
|
|
assert "id" in ev and len(ev["id"]) > 0
|
|
assert "time" in ev
|
|
|
|
|
|
def test_envelope_append(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event, append_event
|
|
ev = make_event("nova.test.event", "run-test-2", "dev", {"key": "value"})
|
|
append_event(ev)
|
|
assert tmp_metrics["events_log"].exists()
|
|
lines = tmp_metrics["events_log"].read_text().strip().split("\n")
|
|
assert len(lines) == 1
|
|
parsed = json.loads(lines[0])
|
|
assert parsed["type"] == "nova.test.event"
|
|
|
|
|
|
# --- Task 2: Per-run manifest writer ---
|
|
|
|
def test_run_manifest_start(tmp_metrics):
|
|
from core.metrics.run_manifest import start_run
|
|
run_id = start_run("cid-123", "dev")
|
|
assert run_id.startswith("run-")
|
|
assert tmp_metrics["events_log"].exists()
|
|
|
|
|
|
def test_run_manifest_complete(tmp_metrics):
|
|
from core.metrics.run_manifest import start_run, complete_run
|
|
run_id = start_run("cid-123", "dev")
|
|
stages = [{"name": "resolve", "duration_ms": 100, "exit_code": 0}]
|
|
manifest = complete_run(run_id, "cid-123", "dev", stages, 0)
|
|
assert manifest["run_id"] == run_id
|
|
assert manifest["exit_code"] == 0
|
|
assert manifest["outcome"] == "succeeded"
|
|
manifest_path = tmp_metrics["runs_dir"] / f"{run_id}.json"
|
|
assert manifest_path.exists()
|
|
saved = json.loads(manifest_path.read_text())
|
|
assert saved["run_id"] == run_id
|
|
|
|
|
|
def test_run_manifest_failed(tmp_metrics):
|
|
from core.metrics.run_manifest import complete_run
|
|
manifest = complete_run("run-fail-1", "cid-123", "dev", [], 1)
|
|
assert manifest["outcome"] == "failed"
|
|
|
|
|
|
# --- Task 6: Decision Ledger (SQLite hash-chain) ---
|
|
|
|
def test_decision_ledger_append(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
from core.metrics.decision_ledger import append, verify_chain
|
|
ev = make_event("nova.ai.decision.made", "run-dl-1", "dev",
|
|
{"decision_id": "run-dl-1", "chosen_action": "pass", "confidence": 0.9,
|
|
"alternatives": {"policy": 1.0}, "human_override": False})
|
|
row = append(ev)
|
|
assert row["seq"] == 1
|
|
assert row["prev_hash"] == "GENESIS"
|
|
ok, broken, _ = verify_chain()
|
|
assert ok
|
|
assert broken == 0
|
|
|
|
|
|
def test_decision_ledger_chain_integrity(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
from core.metrics.decision_ledger import append, verify_chain
|
|
for i in range(5):
|
|
ev = make_event("nova.ai.decision.made", f"run-dl-{i}", "dev",
|
|
{"decision_id": f"run-dl-{i}", "confidence": 0.9 + i * 0.01})
|
|
append(ev)
|
|
ok, broken, details = verify_chain()
|
|
assert ok, f"chain broken: {details}"
|
|
assert broken == 0
|
|
|
|
|
|
def test_decision_ledger_tamper_detection(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
from core.metrics.decision_ledger import append, verify_chain
|
|
ev = make_event("nova.ai.decision.made", "run-tamper-1", "dev", {"confidence": 0.9})
|
|
append(ev)
|
|
# Tamper: directly modify the payload in the DB
|
|
conn = sqlite3.connect(str(tmp_metrics["ledger_db"]))
|
|
conn.execute("UPDATE decision_ledger SET payload = '{}' WHERE seq = 1")
|
|
conn.commit()
|
|
conn.close()
|
|
ok, broken, details = verify_chain()
|
|
assert not ok
|
|
assert broken > 0
|
|
|
|
|
|
def test_decision_ledger_query_by_run(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
from core.metrics.decision_ledger import append, query_by_run
|
|
ev = make_event("nova.ai.decision.made", "run-query-1", "dev", {"confidence": 0.9})
|
|
append(ev)
|
|
entries = query_by_run("run-query-1")
|
|
assert len(entries) == 1
|
|
assert entries[0]["event_type"] == "nova.ai.decision.made"
|
|
|
|
|
|
def test_decision_ledger_stats(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
from core.metrics.decision_ledger import append, stats
|
|
for env in ("dev", "qa", "dev"):
|
|
ev = make_event("nova.ai.decision.made", f"run-stats-{env}", env, {"confidence": 0.9})
|
|
append(ev)
|
|
s = stats()
|
|
assert s["total"] == 3
|
|
assert s["by_environment"].get("dev", 0) == 2
|
|
assert s["by_environment"].get("qa", 0) == 1
|
|
|
|
|
|
def test_decision_ledger_replay(tmp_metrics):
|
|
from core.metrics.event_envelope import make_event
|
|
from core.metrics.decision_ledger import append, replay_run
|
|
ev = make_event("nova.ai.decision.made", "run-replay-1", "dev",
|
|
{"decision_id": "run-replay-1", "chosen_action": "pass",
|
|
"confidence": 0.94, "human_override": False})
|
|
append(ev)
|
|
replay = replay_run("run-replay-1")
|
|
assert "run-replay-1" in replay
|
|
assert "nova.ai.decision.made" in replay
|
|
|
|
|
|
# --- Task 8: Confidence signal event emission ---
|
|
|
|
def test_confidence_event_emission(tmp_metrics):
|
|
from core.confidence_signal import compute
|
|
inputs = {
|
|
"policy": [{"result": "pass", "severity": "info"}],
|
|
"validation": {"schema": True, "stack_resolved": True, "tf_validated": True, "tf_planned": True},
|
|
"freshness": {"age_days": 0, "max_age_days": 1},
|
|
"source": {"submitter": "test", "commit_sha": "abc"},
|
|
"history": {"prior_rollbacks": 0, "prior_policy_fails": 0},
|
|
"nfrs": {"conformance": 1.0},
|
|
}
|
|
sig = compute("cid-conf-1", "dev", inputs)
|
|
assert sig.band == "pass"
|
|
# Check that events were emitted
|
|
assert tmp_metrics["events_log"].exists()
|
|
lines = tmp_metrics["events_log"].read_text().strip().split("\n")
|
|
types = [json.loads(l)["type"] for l in lines]
|
|
assert "nova.confidence.computed" in types
|
|
assert "nova.ai.decision.made" in types
|
|
# Check the decision ledger has the entry
|
|
from core.metrics.decision_ledger import query_by_run
|
|
entries = query_by_run(lines[0].split('"run_id":"')[1].split('"')[0] if '"run_id":"' in lines[0] else "")
|
|
# The run_id is dynamic; just verify the ledger has entries
|
|
from core.metrics.decision_ledger import stats
|
|
s = stats()
|
|
assert s["total"] > 0
|
|
|
|
|
|
# --- Task 7: Attestation event emission ---
|
|
|
|
def test_attestation_event_emission(tmp_metrics):
|
|
from core.hitl_gates import attest
|
|
# Dev skips (autonomous) — no event
|
|
ok, reason = attest("cid-attest-1", "dev", "testuser")
|
|
assert ok
|
|
# QA requires approver + attestation matrix — mock evidence with
|
|
# fresh timestamps (relative to now, not hardcoded — avoids the
|
|
# time-bomb where fixed dates age out of the freshness window).
|
|
import datetime
|
|
now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
ok, reason = attest("cid-attest-2", "qa", "testuser",
|
|
evidence={"functional_correctness": {"timestamp": now, "type": "test", "payload": {}},
|
|
"performance_baseline": {"timestamp": now, "type": "test", "payload": {}},
|
|
"security_posture": {"timestamp": now, "type": "test", "payload": {}},
|
|
"contract_nfrs": {"valid": True}})
|
|
assert ok
|
|
# Check the attestation event was emitted
|
|
from core.metrics.decision_ledger import stats
|
|
s = stats()
|
|
assert s["total"] > 0
|
|
|
|
|
|
# --- Task 9: Policy event emission ---
|
|
|
|
def test_policy_event_emission(tmp_metrics, tmp_path):
|
|
"""Test that checkov_adapter emits nova.policy.evaluated when given a run_id."""
|
|
checkov_json = tmp_path / "checkov.json"
|
|
checkov_json.write_text(json.dumps({
|
|
"terraform_plan": {
|
|
"results": {
|
|
"passed_checks": [{"check_id": "CKV_AWS_1", "check_name": "test", "file_path": "main.tf"}],
|
|
"failed_checks": [],
|
|
"skipped_checks": [],
|
|
}
|
|
}
|
|
}))
|
|
from adapters.terraform.policy.checkov_adapter import adapt
|
|
pcrs = adapt(str(checkov_json), "cid-policy-1", run_id="run-policy-1", environment="dev")
|
|
assert len(pcrs) == 1
|
|
assert pcrs[0]["result"] == "pass"
|
|
# Check the event was emitted
|
|
assert tmp_metrics["events_log"].exists()
|
|
lines = tmp_metrics["events_log"].read_text().strip().split("\n")
|
|
types = [json.loads(l)["type"] for l in lines]
|
|
assert "nova.policy.evaluated" in types
|
|
|
|
|
|
# --- Task 5: Infracost adapter (degraded mode) ---
|
|
|
|
def test_infracost_degraded_mode(tmp_metrics):
|
|
"""When Infracost CLI is absent, the adapter degrades gracefully (A6)."""
|
|
from core.metrics.infracost_adapter import estimate
|
|
# Infracost is not installed in the test env — degraded mode
|
|
result = estimate("/nonexistent/plan.json", "run-infracost-1", "cid-1", "dev")
|
|
assert result["available"] is False
|
|
assert result["delta_usd"] == 0.0
|
|
|
|
|
|
# --- Task 3: Persist ephemeral $WORK/*.json ---
|
|
|
|
def test_persist_run_artifacts(tmp_metrics, tmp_path):
|
|
from core.metrics.run_manifest import persist_run_artifacts
|
|
work_dir = tmp_path / "work"
|
|
work_dir.mkdir()
|
|
(work_dir / "pcr.json").write_text('{"test": true}')
|
|
(work_dir / "signal.json").write_text('{"score": 0.9}')
|
|
copied = persist_run_artifacts("run-persist-1", str(work_dir))
|
|
assert "pcr.json" in copied
|
|
assert "signal.json" in copied
|
|
dest = tmp_metrics["runs_dir"] / "run-persist-1"
|
|
assert (dest / "pcr.json").exists()
|
|
assert (dest / "signal.json").exists() |