generated from coulomb/repo-seed
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
import os
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from phase_memory.credentialed_drills import (
|
|
CREDENTIALED_ADAPTER_ENV_VARS,
|
|
credentialed_adapter_smoke_report,
|
|
credentialed_operator_report,
|
|
credentialed_telemetry_retention_drill,
|
|
missing_credentialed_adapter_env,
|
|
write_credentialed_operator_report,
|
|
)
|
|
|
|
|
|
def test_credentialed_adapter_drill_reports_missing_env_without_secrets() -> None:
|
|
report = credentialed_adapter_smoke_report({})
|
|
|
|
assert report["valid"] is False
|
|
assert report["skipped"] is True
|
|
assert tuple(report["missing_env"]) == CREDENTIALED_ADAPTER_ENV_VARS
|
|
assert report["diagnostics"][0]["code"] == "credential_env_missing"
|
|
|
|
|
|
def test_credentialed_operator_report_redacts_values_and_persists(tmp_path) -> None:
|
|
environ = {
|
|
"PHASE_MEMORY_MARKITECT_URL": "https://markitect.example.invalid",
|
|
"PHASE_MEMORY_MARKITECT_TOKEN": "markitect-secret-token",
|
|
"PHASE_MEMORY_KONTEXTUAL_URL": "https://kontextual.example.invalid",
|
|
"PHASE_MEMORY_KONTEXTUAL_TOKEN": "kontextual-secret-token",
|
|
}
|
|
|
|
report = credentialed_operator_report(environ, run_id="pytest")
|
|
written = write_credentialed_operator_report(tmp_path / "operator-report.json", environ, run_id="pytest")
|
|
serialized = json.dumps(written, sort_keys=True)
|
|
|
|
assert report["valid"] is True
|
|
assert written["id"] == report["id"]
|
|
assert written["redacted_env"]["secrets_redacted"] is True
|
|
assert "markitect-secret-token" not in serialized
|
|
assert "kontextual-secret-token" not in serialized
|
|
assert "https://markitect.example.invalid" not in serialized
|
|
assert "https://kontextual.example.invalid" not in serialized
|
|
assert (tmp_path / "operator-report.json").exists()
|
|
|
|
|
|
def test_credentialed_telemetry_retention_drill_prunes_fixture_events() -> None:
|
|
report = credentialed_telemetry_retention_drill(
|
|
{},
|
|
operator_approved_fixture=True,
|
|
retention_days=30,
|
|
now=datetime(2026, 5, 19, tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert report["valid"] is True
|
|
assert report["skipped"] is False
|
|
assert "op:old" in report["pruned_operation_ids"]
|
|
assert "op:new" in report["retained_operation_ids"]
|
|
assert "audit.retention.apply" in report["audit_operations"]
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
missing_credentialed_adapter_env(os.environ),
|
|
reason="requires env vars: " + ", ".join(CREDENTIALED_ADAPTER_ENV_VARS),
|
|
)
|
|
def test_credentialed_adapter_drill_reuses_manifest_contract_when_env_is_present() -> None:
|
|
report = credentialed_adapter_smoke_report(os.environ)
|
|
|
|
assert report["valid"] is True
|
|
assert report["skipped"] is False
|
|
assert report["adapter_pack"]["name"] == "live-shaped"
|
|
assert report["config"]["credential_fingerprint"]
|
|
assert "PHASE_MEMORY_MARKITECT_TOKEN" not in str(report)
|