402 lines
13 KiB
Python
402 lines
13 KiB
Python
"""Tests for health_snapshot module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
# Add timmy_automations to path for imports
|
|
sys.path.insert(
|
|
0, str(Path(__file__).resolve().parent.parent.parent / "timmy_automations" / "daily_run")
|
|
)
|
|
|
|
from datetime import UTC
|
|
|
|
import health_snapshot as hs
|
|
|
|
|
|
class TestLoadConfig:
|
|
"""Test configuration loading."""
|
|
|
|
def test_loads_default_config(self):
|
|
"""Load default configuration."""
|
|
config = hs.load_config()
|
|
|
|
assert "gitea_api" in config
|
|
assert "repo_slug" in config
|
|
assert "critical_labels" in config
|
|
assert "flakiness_lookback_cycles" in config
|
|
|
|
def test_environment_overrides(self, monkeypatch):
|
|
"""Environment variables override defaults."""
|
|
monkeypatch.setenv("TIMMY_GITEA_API", "http://test:3000/api/v1")
|
|
monkeypatch.setenv("TIMMY_REPO_SLUG", "test/repo")
|
|
|
|
config = hs.load_config()
|
|
|
|
assert config["gitea_api"] == "http://test:3000/api/v1"
|
|
assert config["repo_slug"] == "test/repo"
|
|
|
|
|
|
class TestGetToken:
|
|
"""Test token retrieval."""
|
|
|
|
def test_returns_config_token(self):
|
|
"""Return token from config if present."""
|
|
config = {"token": "test-token-123"}
|
|
token = hs.get_token(config)
|
|
|
|
assert token == "test-token-123"
|
|
|
|
def test_reads_from_file(self, tmp_path, monkeypatch):
|
|
"""Read token from file if no config token."""
|
|
token_file = tmp_path / "gitea_token"
|
|
token_file.write_text("file-token-456")
|
|
|
|
config = {"token_file": str(token_file)}
|
|
token = hs.get_token(config)
|
|
|
|
assert token == "file-token-456"
|
|
|
|
def test_returns_none_when_no_token(self):
|
|
"""Return None when no token available."""
|
|
config = {"token_file": "/nonexistent/path"}
|
|
token = hs.get_token(config)
|
|
|
|
assert token is None
|
|
|
|
|
|
class TestCISignal:
|
|
"""Test CISignal dataclass."""
|
|
|
|
def test_default_details(self):
|
|
"""Details defaults to empty dict."""
|
|
signal = hs.CISignal(status="pass", message="CI passing")
|
|
|
|
assert signal.details == {}
|
|
|
|
def test_with_details(self):
|
|
"""Can include details."""
|
|
signal = hs.CISignal(status="pass", message="CI passing", details={"sha": "abc123"})
|
|
|
|
assert signal.details["sha"] == "abc123"
|
|
|
|
|
|
class TestIssueSignal:
|
|
"""Test IssueSignal dataclass."""
|
|
|
|
def test_default_issues_list(self):
|
|
"""Issues defaults to empty list."""
|
|
signal = hs.IssueSignal(count=0, p0_count=0, p1_count=0)
|
|
|
|
assert signal.issues == []
|
|
|
|
def test_with_issues(self):
|
|
"""Can include issues."""
|
|
issues = [{"number": 1, "title": "Test"}]
|
|
signal = hs.IssueSignal(count=1, p0_count=1, p1_count=0, issues=issues)
|
|
|
|
assert len(signal.issues) == 1
|
|
|
|
|
|
class TestFlakinessSignal:
|
|
"""Test FlakinessSignal dataclass."""
|
|
|
|
def test_calculated_fields(self):
|
|
"""All fields set correctly."""
|
|
signal = hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=2,
|
|
recent_cycles=20,
|
|
failure_rate=0.1,
|
|
message="Low flakiness",
|
|
)
|
|
|
|
assert signal.status == "healthy"
|
|
assert signal.recent_failures == 2
|
|
assert signal.failure_rate == 0.1
|
|
|
|
|
|
class TestHealthSnapshot:
|
|
"""Test HealthSnapshot dataclass."""
|
|
|
|
def test_to_dict_structure(self):
|
|
"""to_dict produces expected structure."""
|
|
snapshot = hs.HealthSnapshot(
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
overall_status="green",
|
|
ci=hs.CISignal(status="pass", message="CI passing"),
|
|
issues=hs.IssueSignal(count=0, p0_count=0, p1_count=0),
|
|
flakiness=hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
),
|
|
tokens=hs.TokenEconomySignal(status="balanced", message="Balanced"),
|
|
)
|
|
|
|
data = snapshot.to_dict()
|
|
|
|
assert data["timestamp"] == "2026-01-01T00:00:00+00:00"
|
|
assert data["overall_status"] == "green"
|
|
assert "ci" in data
|
|
assert "issues" in data
|
|
assert "flakiness" in data
|
|
assert "tokens" in data
|
|
|
|
def test_to_dict_limits_issues(self):
|
|
"""to_dict limits issues to 5."""
|
|
many_issues = [{"number": i, "title": f"Issue {i}"} for i in range(10)]
|
|
snapshot = hs.HealthSnapshot(
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
overall_status="green",
|
|
ci=hs.CISignal(status="pass", message="CI passing"),
|
|
issues=hs.IssueSignal(count=10, p0_count=5, p1_count=5, issues=many_issues),
|
|
flakiness=hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
),
|
|
tokens=hs.TokenEconomySignal(status="balanced", message="Balanced"),
|
|
)
|
|
|
|
data = snapshot.to_dict()
|
|
|
|
assert len(data["issues"]["issues"]) == 5
|
|
|
|
|
|
class TestCalculateOverallStatus:
|
|
"""Test overall status calculation."""
|
|
|
|
def test_green_when_all_healthy(self):
|
|
"""Status is green when all signals healthy."""
|
|
ci = hs.CISignal(status="pass", message="CI passing")
|
|
issues = hs.IssueSignal(count=0, p0_count=0, p1_count=0)
|
|
flakiness = hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
)
|
|
|
|
status = hs.calculate_overall_status(ci, issues, flakiness)
|
|
|
|
assert status == "green"
|
|
|
|
def test_red_when_ci_fails(self):
|
|
"""Status is red when CI fails."""
|
|
ci = hs.CISignal(status="fail", message="CI failed")
|
|
issues = hs.IssueSignal(count=0, p0_count=0, p1_count=0)
|
|
flakiness = hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
)
|
|
|
|
status = hs.calculate_overall_status(ci, issues, flakiness)
|
|
|
|
assert status == "red"
|
|
|
|
def test_red_when_p0_issues(self):
|
|
"""Status is red when P0 issues exist."""
|
|
ci = hs.CISignal(status="pass", message="CI passing")
|
|
issues = hs.IssueSignal(count=1, p0_count=1, p1_count=0)
|
|
flakiness = hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
)
|
|
|
|
status = hs.calculate_overall_status(ci, issues, flakiness)
|
|
|
|
assert status == "red"
|
|
|
|
def test_yellow_when_p1_issues(self):
|
|
"""Status is yellow when P1 issues exist."""
|
|
ci = hs.CISignal(status="pass", message="CI passing")
|
|
issues = hs.IssueSignal(count=1, p0_count=0, p1_count=1)
|
|
flakiness = hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
)
|
|
|
|
status = hs.calculate_overall_status(ci, issues, flakiness)
|
|
|
|
assert status == "yellow"
|
|
|
|
def test_yellow_when_flakiness_degraded(self):
|
|
"""Status is yellow when flakiness degraded."""
|
|
ci = hs.CISignal(status="pass", message="CI passing")
|
|
issues = hs.IssueSignal(count=0, p0_count=0, p1_count=0)
|
|
flakiness = hs.FlakinessSignal(
|
|
status="degraded",
|
|
recent_failures=5,
|
|
recent_cycles=20,
|
|
failure_rate=0.25,
|
|
message="Moderate flakiness",
|
|
)
|
|
|
|
status = hs.calculate_overall_status(ci, issues, flakiness)
|
|
|
|
assert status == "yellow"
|
|
|
|
def test_red_when_flakiness_critical(self):
|
|
"""Status is red when flakiness critical."""
|
|
ci = hs.CISignal(status="pass", message="CI passing")
|
|
issues = hs.IssueSignal(count=0, p0_count=0, p1_count=0)
|
|
flakiness = hs.FlakinessSignal(
|
|
status="critical",
|
|
recent_failures=10,
|
|
recent_cycles=20,
|
|
failure_rate=0.5,
|
|
message="High flakiness",
|
|
)
|
|
|
|
status = hs.calculate_overall_status(ci, issues, flakiness)
|
|
|
|
assert status == "red"
|
|
|
|
|
|
class TestCheckFlakiness:
|
|
"""Test flakiness checking."""
|
|
|
|
def test_no_data_returns_unknown(self, tmp_path, monkeypatch):
|
|
"""Return unknown when no cycle data exists."""
|
|
monkeypatch.setattr(hs, "REPO_ROOT", tmp_path)
|
|
config = {"flakiness_lookback_cycles": 20}
|
|
|
|
signal = hs.check_flakiness(config)
|
|
|
|
assert signal.status == "unknown"
|
|
assert signal.message == "No cycle data available"
|
|
|
|
def test_calculates_failure_rate(self, tmp_path, monkeypatch):
|
|
"""Calculate failure rate from cycle data."""
|
|
monkeypatch.setattr(hs, "REPO_ROOT", tmp_path)
|
|
|
|
retro_dir = tmp_path / ".loop" / "retro"
|
|
retro_dir.mkdir(parents=True)
|
|
|
|
cycles = [
|
|
json.dumps({"success": True, "cycle": 1}),
|
|
json.dumps({"success": True, "cycle": 2}),
|
|
json.dumps({"success": False, "cycle": 3}),
|
|
json.dumps({"success": True, "cycle": 4}),
|
|
json.dumps({"success": False, "cycle": 5}),
|
|
]
|
|
retro_file = retro_dir / "cycles.jsonl"
|
|
retro_file.write_text("\n".join(cycles))
|
|
|
|
config = {"flakiness_lookback_cycles": 20}
|
|
signal = hs.check_flakiness(config)
|
|
|
|
assert signal.recent_cycles == 5
|
|
assert signal.recent_failures == 2
|
|
assert signal.failure_rate == 0.4
|
|
assert signal.status == "critical" # 40% > 30%
|
|
|
|
|
|
class TestCheckTokenEconomy:
|
|
"""Test token economy checking."""
|
|
|
|
def test_no_data_returns_unknown(self, tmp_path, monkeypatch):
|
|
"""Return unknown when no token data exists."""
|
|
monkeypatch.setattr(hs, "REPO_ROOT", tmp_path)
|
|
config = {}
|
|
|
|
signal = hs.check_token_economy(config)
|
|
|
|
assert signal.status == "unknown"
|
|
|
|
def test_calculates_balanced(self, tmp_path, monkeypatch):
|
|
"""Detect balanced token economy."""
|
|
monkeypatch.setattr(hs, "REPO_ROOT", tmp_path)
|
|
|
|
loop_dir = tmp_path / ".loop"
|
|
loop_dir.mkdir(parents=True)
|
|
|
|
from datetime import datetime
|
|
|
|
now = datetime.now(UTC).isoformat()
|
|
transactions = [
|
|
json.dumps({"timestamp": now, "delta": 10}),
|
|
json.dumps({"timestamp": now, "delta": -5}),
|
|
]
|
|
ledger_file = loop_dir / "token_economy.jsonl"
|
|
ledger_file.write_text("\n".join(transactions))
|
|
|
|
config = {}
|
|
signal = hs.check_token_economy(config)
|
|
|
|
assert signal.status == "balanced"
|
|
assert signal.recent_mint == 10
|
|
assert signal.recent_burn == 5
|
|
|
|
|
|
class TestGiteaClient:
|
|
"""Test Gitea API client."""
|
|
|
|
def test_initialization(self):
|
|
"""Initialize with config and token."""
|
|
config = {"gitea_api": "http://test:3000/api/v1", "repo_slug": "test/repo"}
|
|
client = hs.GiteaClient(config, "token123")
|
|
|
|
assert client.api_base == "http://test:3000/api/v1"
|
|
assert client.repo_slug == "test/repo"
|
|
assert client.token == "token123"
|
|
|
|
def test_headers_with_token(self):
|
|
"""Include authorization header with token."""
|
|
config = {"gitea_api": "http://test:3000/api/v1", "repo_slug": "test/repo"}
|
|
client = hs.GiteaClient(config, "token123")
|
|
|
|
headers = client._headers()
|
|
|
|
assert headers["Authorization"] == "token token123"
|
|
assert headers["Accept"] == "application/json"
|
|
|
|
def test_headers_without_token(self):
|
|
"""No authorization header without token."""
|
|
config = {"gitea_api": "http://test:3000/api/v1", "repo_slug": "test/repo"}
|
|
client = hs.GiteaClient(config, None)
|
|
|
|
headers = client._headers()
|
|
|
|
assert "Authorization" not in headers
|
|
assert headers["Accept"] == "application/json"
|
|
|
|
|
|
class TestGenerateSnapshot:
|
|
"""Test snapshot generation."""
|
|
|
|
def test_returns_snapshot(self):
|
|
"""Generate a complete snapshot."""
|
|
config = hs.load_config()
|
|
|
|
with (
|
|
patch.object(hs.GiteaClient, "is_available", return_value=False),
|
|
patch.object(hs.GiteaClient, "__init__", return_value=None),
|
|
):
|
|
snapshot = hs.generate_snapshot(config, None)
|
|
|
|
assert isinstance(snapshot, hs.HealthSnapshot)
|
|
assert snapshot.overall_status in ["green", "yellow", "red", "unknown"]
|
|
assert snapshot.ci is not None
|
|
assert snapshot.issues is not None
|
|
assert snapshot.flakiness is not None
|
|
assert snapshot.tokens is not None
|