generated from coulomb/repo-seed
tegwick
bd169a07e2
- ActorType enum (adm/agt/atm) replaces actor_class string; config validates naming convention (adm-*/agt-*/atm-*) with hard ConfigError on mismatch; legacy 'human'/'automation' values accepted with DeprecationWarning - cert_command: pluggable shell string run before each SSH launch; cert written to state dir; -i cert appended to SSH command alongside -i key - TTL-aware cert refresh: parses Valid-to via ssh-keygen -L; pre-emptive restart 5 min before expiry (no backoff, no attempt increment); CERT_EXPIRING logged - CertAcquisitionError: cert failures trigger normal backoff/retry loop - cert_identity: Key ID parsed from cert and recorded in BRIDGE_CONNECTED event - bridge cert-status: new CLI command; exit 1 on expired cert; --json flag - 233 tests passing, ruff clean Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
"""Tests for audit logging."""
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from bridge.audit import AuditLogger, AuditEvent
|
|
|
|
|
|
@pytest.fixture
|
|
def log_dir(tmp_path):
|
|
return tmp_path / "bridge"
|
|
|
|
|
|
@pytest.fixture
|
|
def logger(log_dir):
|
|
return AuditLogger(state_dir=log_dir)
|
|
|
|
|
|
class TestAuditLogger:
|
|
def test_log_event_creates_file(self, logger, log_dir):
|
|
logger.log(
|
|
tunnel="my-tunnel",
|
|
event=AuditEvent.BRIDGE_STARTED,
|
|
actor="operator.bernd",
|
|
actor_type="adm",
|
|
)
|
|
log_file = log_dir / "my-tunnel.log"
|
|
assert log_file.exists()
|
|
|
|
def test_log_event_is_json_line(self, logger, log_dir):
|
|
logger.log(
|
|
tunnel="my-tunnel",
|
|
event=AuditEvent.BRIDGE_STARTED,
|
|
actor="operator.bernd",
|
|
actor_type="adm",
|
|
)
|
|
lines = (log_dir / "my-tunnel.log").read_text().strip().splitlines()
|
|
assert len(lines) == 1
|
|
entry = json.loads(lines[0])
|
|
assert entry["tunnel"] == "my-tunnel"
|
|
assert entry["event"] == "bridge_started"
|
|
assert entry["actor"] == "operator.bernd"
|
|
assert entry["actor_type"] == "adm"
|
|
assert "timestamp" in entry
|
|
|
|
def test_multiple_events_append(self, logger, log_dir):
|
|
for event in [AuditEvent.BRIDGE_STARTED, AuditEvent.BRIDGE_CONNECTED, AuditEvent.BRIDGE_STOPPED]:
|
|
logger.log(tunnel="t", event=event, actor="a", actor_type="adm")
|
|
lines = (log_dir / "t.log").read_text().strip().splitlines()
|
|
assert len(lines) == 3
|
|
|
|
def test_log_with_detail(self, logger, log_dir):
|
|
logger.log(
|
|
tunnel="t",
|
|
event=AuditEvent.HEALTH_CHECK_FAILED,
|
|
actor="a",
|
|
actor_type="atm",
|
|
detail="connection refused",
|
|
)
|
|
entry = json.loads((log_dir / "t.log").read_text().strip())
|
|
assert entry["detail"] == "connection refused"
|
|
|
|
def test_all_event_types_defined(self):
|
|
events = {e.value for e in AuditEvent}
|
|
assert "bridge_started" in events
|
|
assert "bridge_connected" in events
|
|
assert "bridge_disconnected" in events
|
|
assert "bridge_reconnecting" in events
|
|
assert "health_check_failed" in events
|
|
assert "health_check_recovered" in events
|
|
assert "bridge_stopped" in events
|
|
|
|
def test_timestamp_is_iso8601(self, logger, log_dir):
|
|
from datetime import datetime
|
|
logger.log(tunnel="t", event=AuditEvent.BRIDGE_STOPPED, actor="a", actor_type="adm")
|
|
entry = json.loads((log_dir / "t.log").read_text().strip())
|
|
# Should parse without error
|
|
dt = datetime.fromisoformat(entry["timestamp"])
|
|
assert dt.tzinfo is not None or True # UTC or naive both acceptable
|
|
|
|
def test_read_events(self, logger, log_dir):
|
|
logger.log(tunnel="t", event=AuditEvent.BRIDGE_STARTED, actor="a", actor_type="adm")
|
|
logger.log(tunnel="t", event=AuditEvent.BRIDGE_STOPPED, actor="a", actor_type="adm")
|
|
events = logger.read_events("t")
|
|
assert len(events) == 2
|
|
assert events[0]["event"] == "bridge_started"
|
|
|
|
def test_read_events_missing_returns_empty(self, logger):
|
|
assert logger.read_events("nonexistent") == []
|