344 lines
12 KiB
Python
344 lines
12 KiB
Python
"""Tests for weekly_narrative.py script."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Add timmy_automations to path for imports
|
|
sys.path.insert(
|
|
0, str(Path(__file__).resolve().parent.parent.parent / "timmy_automations" / "daily_run")
|
|
)
|
|
|
|
import weekly_narrative as wn
|
|
|
|
|
|
class TestParseTimestamp:
|
|
"""Test timestamp parsing."""
|
|
|
|
def test_parse_iso_with_z(self):
|
|
"""Parse ISO timestamp with Z suffix."""
|
|
result = wn.parse_ts("2026-03-21T12:00:00Z")
|
|
assert result is not None
|
|
assert result.year == 2026
|
|
assert result.month == 3
|
|
assert result.day == 21
|
|
|
|
def test_parse_iso_with_offset(self):
|
|
"""Parse ISO timestamp with timezone offset."""
|
|
result = wn.parse_ts("2026-03-21T12:00:00+00:00")
|
|
assert result is not None
|
|
assert result.year == 2026
|
|
|
|
def test_parse_empty_string(self):
|
|
"""Empty string returns None."""
|
|
result = wn.parse_ts("")
|
|
assert result is None
|
|
|
|
def test_parse_invalid_string(self):
|
|
"""Invalid string returns None."""
|
|
result = wn.parse_ts("not-a-timestamp")
|
|
assert result is None
|
|
|
|
|
|
class TestCollectCyclesData:
|
|
"""Test cycle data collection."""
|
|
|
|
def test_no_cycles_file(self, tmp_path):
|
|
"""Handle missing cycles file gracefully."""
|
|
with patch.object(wn, "REPO_ROOT", tmp_path):
|
|
since = datetime.now(UTC) - timedelta(days=7)
|
|
result = wn.collect_cycles_data(since)
|
|
assert result["total"] == 0
|
|
assert result["successes"] == 0
|
|
assert result["failures"] == 0
|
|
|
|
def test_collect_recent_cycles(self, tmp_path):
|
|
"""Collect cycles within lookback period."""
|
|
retro_dir = tmp_path / ".loop" / "retro"
|
|
retro_dir.mkdir(parents=True)
|
|
|
|
now = datetime.now(UTC)
|
|
cycles = [
|
|
{"timestamp": now.isoformat(), "success": True, "cycle": 1},
|
|
{"timestamp": now.isoformat(), "success": False, "cycle": 2},
|
|
{"timestamp": (now - timedelta(days=10)).isoformat(), "success": True, "cycle": 3},
|
|
]
|
|
|
|
with open(retro_dir / "cycles.jsonl", "w") as f:
|
|
for c in cycles:
|
|
f.write(json.dumps(c) + "\n")
|
|
|
|
with patch.object(wn, "REPO_ROOT", tmp_path):
|
|
since = now - timedelta(days=7)
|
|
result = wn.collect_cycles_data(since)
|
|
assert result["total"] == 2 # Only recent 2
|
|
assert result["successes"] == 1
|
|
assert result["failures"] == 1
|
|
|
|
|
|
class TestExtractThemes:
|
|
"""Test theme extraction from issues."""
|
|
|
|
def test_extract_layer_labels(self):
|
|
"""Extract layer labels from issues."""
|
|
issues = [
|
|
{"labels": [{"name": "layer:triage"}, {"name": "bug"}]},
|
|
{"labels": [{"name": "layer:tests"}, {"name": "bug"}]},
|
|
{"labels": [{"name": "layer:triage"}, {"name": "feature"}]},
|
|
]
|
|
|
|
result = wn.extract_themes(issues)
|
|
|
|
assert len(result["layers"]) == 2
|
|
layer_names = {layer["name"] for layer in result["layers"]}
|
|
assert "triage" in layer_names
|
|
assert "tests" in layer_names
|
|
|
|
def test_extract_type_labels(self):
|
|
"""Extract type labels (bug/feature/etc)."""
|
|
issues = [
|
|
{"labels": [{"name": "bug"}]},
|
|
{"labels": [{"name": "feature"}]},
|
|
{"labels": [{"name": "bug"}]},
|
|
]
|
|
|
|
result = wn.extract_themes(issues)
|
|
|
|
type_names = {t_type["name"] for t_type in result["types"]}
|
|
assert "bug" in type_names
|
|
assert "feature" in type_names
|
|
|
|
def test_empty_issues(self):
|
|
"""Handle empty issue list."""
|
|
result = wn.extract_themes([])
|
|
assert result["layers"] == []
|
|
assert result["types"] == []
|
|
assert result["top_labels"] == []
|
|
|
|
|
|
class TestExtractAgentContributions:
|
|
"""Test agent contribution extraction."""
|
|
|
|
def test_extract_assignees(self):
|
|
"""Extract assignee counts."""
|
|
issues = [
|
|
{"assignee": {"login": "kimi"}},
|
|
{"assignee": {"login": "hermes"}},
|
|
{"assignee": {"login": "kimi"}},
|
|
]
|
|
|
|
result = wn.extract_agent_contributions(issues, [], [])
|
|
|
|
assert len(result["active_assignees"]) == 2
|
|
assignee_logins = {a["login"] for a in result["active_assignees"]} # noqa: E741
|
|
assert "kimi" in assignee_logins
|
|
assert "hermes" in assignee_logins
|
|
|
|
def test_extract_pr_authors(self):
|
|
"""Extract PR author counts."""
|
|
prs = [
|
|
{"user": {"login": "kimi"}},
|
|
{"user": {"login": "claude"}},
|
|
{"user": {"login": "kimi"}},
|
|
]
|
|
|
|
result = wn.extract_agent_contributions([], prs, [])
|
|
|
|
assert len(result["pr_authors"]) == 2
|
|
|
|
def test_kimi_mentions_in_cycles(self):
|
|
"""Count Kimi mentions in cycle notes."""
|
|
cycles = [
|
|
{"notes": "Kimi did great work", "reason": ""},
|
|
{"notes": "", "reason": "Kimi timeout"},
|
|
{"notes": "All good", "reason": ""},
|
|
]
|
|
|
|
result = wn.extract_agent_contributions([], [], cycles)
|
|
assert result["kimi_mentioned_cycles"] == 2
|
|
|
|
|
|
class TestAnalyzeTestShifts:
|
|
"""Test test pattern analysis."""
|
|
|
|
def test_no_cycles(self):
|
|
"""Handle no cycle data."""
|
|
result = wn.analyze_test_shifts([])
|
|
assert "note" in result
|
|
|
|
def test_test_metrics(self):
|
|
"""Calculate test metrics from cycles."""
|
|
cycles = [
|
|
{"tests_passed": 100, "tests_added": 5},
|
|
{"tests_passed": 150, "tests_added": 3},
|
|
]
|
|
|
|
result = wn.analyze_test_shifts(cycles)
|
|
|
|
assert result["total_tests_passed"] == 250
|
|
assert result["total_tests_added"] == 8
|
|
|
|
|
|
class TestGenerateVibeSummary:
|
|
"""Test vibe summary generation."""
|
|
|
|
def test_productive_vibe(self):
|
|
"""High success rate and activity = productive vibe."""
|
|
cycles_data = {"success_rate": 0.95, "successes": 10, "failures": 1}
|
|
issues_data = {"closed_count": 5}
|
|
|
|
result = wn.generate_vibe_summary(cycles_data, issues_data, {}, {"layers": []}, {}, {}, {})
|
|
|
|
assert result["overall"] == "productive"
|
|
assert "strong week" in result["description"].lower()
|
|
|
|
def test_struggling_vibe(self):
|
|
"""More failures than successes = struggling vibe."""
|
|
cycles_data = {"success_rate": 0.3, "successes": 3, "failures": 7}
|
|
issues_data = {"closed_count": 0}
|
|
|
|
result = wn.generate_vibe_summary(cycles_data, issues_data, {}, {"layers": []}, {}, {}, {})
|
|
|
|
assert result["overall"] == "struggling"
|
|
|
|
def test_quiet_vibe(self):
|
|
"""Low activity = quiet vibe."""
|
|
cycles_data = {"success_rate": 0.0, "successes": 0, "failures": 0}
|
|
issues_data = {"closed_count": 0}
|
|
|
|
result = wn.generate_vibe_summary(cycles_data, issues_data, {}, {"layers": []}, {}, {}, {})
|
|
|
|
assert result["overall"] == "quiet"
|
|
|
|
|
|
class TestGenerateMarkdownSummary:
|
|
"""Test markdown summary generation."""
|
|
|
|
def test_includes_header(self):
|
|
"""Markdown includes header."""
|
|
narrative = {
|
|
"period": {"start": "2026-03-14T00:00:00", "end": "2026-03-21T00:00:00"},
|
|
"vibe": {"overall": "productive", "description": "Good week"},
|
|
"activity": {
|
|
"cycles": {"total": 10, "successes": 9, "failures": 1},
|
|
"issues": {"closed": 5, "opened": 3},
|
|
"pull_requests": {"merged": 4, "opened": 2},
|
|
},
|
|
}
|
|
|
|
result = wn.generate_markdown_summary(narrative)
|
|
|
|
assert "# Weekly Narrative Summary" in result
|
|
assert "productive" in result.lower()
|
|
assert "10 total" in result or "10" in result
|
|
|
|
def test_includes_focus_areas(self):
|
|
"""Markdown includes focus areas when present."""
|
|
narrative = {
|
|
"period": {"start": "2026-03-14", "end": "2026-03-21"},
|
|
"vibe": {
|
|
"overall": "productive",
|
|
"description": "Good week",
|
|
"focus_areas": ["triage (5 items)", "tests (3 items)"],
|
|
},
|
|
"activity": {
|
|
"cycles": {"total": 0, "successes": 0, "failures": 0},
|
|
"issues": {"closed": 0, "opened": 0},
|
|
"pull_requests": {"merged": 0, "opened": 0},
|
|
},
|
|
}
|
|
|
|
result = wn.generate_markdown_summary(narrative)
|
|
|
|
assert "Focus Areas" in result
|
|
assert "triage" in result
|
|
|
|
|
|
class TestConfigLoading:
|
|
"""Test configuration loading."""
|
|
|
|
def test_default_config(self, tmp_path):
|
|
"""Default config when manifest missing."""
|
|
with patch.object(wn, "CONFIG_PATH", tmp_path / "nonexistent.json"):
|
|
config = wn.load_automation_config()
|
|
assert config["lookback_days"] == 7
|
|
assert config["enabled"] is True
|
|
|
|
def test_environment_override(self, tmp_path):
|
|
"""Environment variables override config."""
|
|
with patch.dict("os.environ", {"TIMMY_WEEKLY_NARRATIVE_ENABLED": "false"}):
|
|
with patch.object(wn, "CONFIG_PATH", tmp_path / "nonexistent.json"):
|
|
config = wn.load_automation_config()
|
|
assert config["enabled"] is False
|
|
|
|
|
|
class TestMain:
|
|
"""Test main function."""
|
|
|
|
def test_disabled_exits_cleanly(self, tmp_path):
|
|
"""When disabled and no --force, exits cleanly."""
|
|
with patch.object(wn, "REPO_ROOT", tmp_path):
|
|
with patch.object(wn, "load_automation_config", return_value={"enabled": False}):
|
|
with patch("sys.argv", ["weekly_narrative"]):
|
|
result = wn.main()
|
|
assert result == 0
|
|
|
|
def test_force_runs_when_disabled(self, tmp_path):
|
|
"""--force runs even when disabled."""
|
|
# Setup minimal structure
|
|
(tmp_path / ".loop" / "retro").mkdir(parents=True)
|
|
|
|
with patch.object(wn, "REPO_ROOT", tmp_path):
|
|
with patch.object(
|
|
wn,
|
|
"load_automation_config",
|
|
return_value={
|
|
"enabled": False,
|
|
"lookback_days": 7,
|
|
"gitea_api": "http://localhost:3000/api/v1",
|
|
"repo_slug": "test/repo",
|
|
"token_file": "~/.hermes/gitea_token",
|
|
},
|
|
):
|
|
with patch.object(wn, "GiteaClient") as mock_client:
|
|
mock_instance = MagicMock()
|
|
mock_instance.is_available.return_value = False
|
|
mock_client.return_value = mock_instance
|
|
|
|
with patch("sys.argv", ["weekly_narrative", "--force"]):
|
|
result = wn.main()
|
|
# Should complete without error even though Gitea unavailable
|
|
assert result == 0
|
|
|
|
|
|
class TestGiteaClient:
|
|
"""Test Gitea API client."""
|
|
|
|
def test_is_available_when_unavailable(self):
|
|
"""is_available returns False when server down."""
|
|
config = {"gitea_api": "http://localhost:99999", "repo_slug": "test/repo"}
|
|
client = wn.GiteaClient(config, None)
|
|
|
|
# Should return False without raising
|
|
assert client.is_available() is False
|
|
|
|
def test_headers_with_token(self):
|
|
"""Headers include Authorization when token provided."""
|
|
config = {"gitea_api": "http://localhost:3000", "repo_slug": "test/repo"}
|
|
client = wn.GiteaClient(config, "test-token")
|
|
|
|
headers = client._headers()
|
|
assert headers["Authorization"] == "token test-token"
|
|
|
|
def test_headers_without_token(self):
|
|
"""Headers don't include Authorization when no token."""
|
|
config = {"gitea_api": "http://localhost:3000", "repo_slug": "test/repo"}
|
|
client = wn.GiteaClient(config, None)
|
|
|
|
headers = client._headers()
|
|
assert "Authorization" not in headers
|