[claude] feat: Session Sovereignty Report Generator (#957) v3 (#1263)

Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>

tests/timmy/test_session_report.py

@@ -0,0 +1,444 @@
+"""Tests for timmy.sovereignty.session_report.
+
+Refs: #957 (Session Sovereignty Report Generator)
+"""
+
+import base64
+import json
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+from timmy.sovereignty.session_report import (
+    _format_duration,
+    _gather_session_data,
+    _gather_sovereignty_data,
+    _render_markdown,
+    commit_report,
+    generate_and_commit_report,
+    generate_report,
+    mark_session_start,
+)
+
+
+# ---------------------------------------------------------------------------
+# _format_duration
+# ---------------------------------------------------------------------------
+
+
+class TestFormatDuration:
+    def test_seconds_only(self):
+        assert _format_duration(45) == "45s"
+
+    def test_minutes_and_seconds(self):
+        assert _format_duration(125) == "2m 5s"
+
+    def test_hours_minutes_seconds(self):
+        assert _format_duration(3661) == "1h 1m 1s"
+
+    def test_zero(self):
+        assert _format_duration(0) == "0s"
+
+
+# ---------------------------------------------------------------------------
+# mark_session_start + generate_report (smoke)
+# ---------------------------------------------------------------------------
+
+
+class TestMarkSessionStart:
+    def test_sets_session_start(self):
+        import timmy.sovereignty.session_report as sr
+
+        sr._SESSION_START = None
+        mark_session_start()
+        assert sr._SESSION_START is not None
+        assert sr._SESSION_START.tzinfo == UTC
+
+    def test_idempotent_overwrite(self):
+        import timmy.sovereignty.session_report as sr
+
+        mark_session_start()
+        first = sr._SESSION_START
+        time.sleep(0.01)
+        mark_session_start()
+        second = sr._SESSION_START
+        assert second >= first
+
+
+# ---------------------------------------------------------------------------
+# _gather_session_data
+# ---------------------------------------------------------------------------
+
+
+class TestGatherSessionData:
+    def test_returns_defaults_when_no_file(self, tmp_path):
+        mock_logger = MagicMock()
+        mock_logger.flush.return_value = None
+        mock_logger.session_file = tmp_path / "nonexistent.jsonl"
+
+        with patch(
+            "timmy.sovereignty.session_report.get_session_logger",
+            return_value=mock_logger,
+        ):
+            data = _gather_session_data()
+
+        assert data["user_messages"] == 0
+        assert data["timmy_messages"] == 0
+        assert data["tool_calls"] == 0
+        assert data["errors"] == 0
+        assert data["tool_call_breakdown"] == {}
+
+    def test_counts_entries_correctly(self, tmp_path):
+        session_file = tmp_path / "session_2026-03-23.jsonl"
+        entries = [
+            {"type": "message", "role": "user", "content": "hello"},
+            {"type": "message", "role": "timmy", "content": "hi"},
+            {"type": "message", "role": "user", "content": "test"},
+            {"type": "tool_call", "tool": "memory_search", "args": {}, "result": "found"},
+            {"type": "tool_call", "tool": "memory_search", "args": {}, "result": "nope"},
+            {"type": "tool_call", "tool": "shell", "args": {}, "result": "ok"},
+            {"type": "error", "error": "boom"},
+        ]
+        with open(session_file, "w") as f:
+            for e in entries:
+                f.write(json.dumps(e) + "\n")
+
+        mock_logger = MagicMock()
+        mock_logger.flush.return_value = None
+        mock_logger.session_file = session_file
+
+        with patch(
+            "timmy.sovereignty.session_report.get_session_logger",
+            return_value=mock_logger,
+        ):
+            data = _gather_session_data()
+
+        assert data["user_messages"] == 2
+        assert data["timmy_messages"] == 1
+        assert data["tool_calls"] == 3
+        assert data["errors"] == 1
+        assert data["tool_call_breakdown"]["memory_search"] == 2
+        assert data["tool_call_breakdown"]["shell"] == 1
+
+    def test_graceful_on_import_error(self):
+        with patch(
+            "timmy.sovereignty.session_report.get_session_logger",
+            side_effect=ImportError("no session_logger"),
+        ):
+            data = _gather_session_data()
+
+        assert data["tool_calls"] == 0
+
+
+# ---------------------------------------------------------------------------
+# _gather_sovereignty_data
+# ---------------------------------------------------------------------------
+
+
+class TestGatherSovereigntyData:
+    def test_returns_empty_on_import_error(self):
+        with patch.dict("sys.modules", {"infrastructure.sovereignty_metrics": None}):
+            with patch(
+                "timmy.sovereignty.session_report.get_sovereignty_store",
+                side_effect=ImportError("no store"),
+            ):
+                data = _gather_sovereignty_data()
+
+        assert data["metrics"] == {}
+        assert data["deltas"] == {}
+        assert data["previous_session"] == {}
+
+    def test_populates_deltas_from_history(self):
+        mock_store = MagicMock()
+        mock_store.get_summary.return_value = {
+            "cache_hit_rate": {"current": 0.5, "phase": "week1"},
+        }
+        # get_latest returns newest-first
+        mock_store.get_latest.return_value = [
+            {"value": 0.5},
+            {"value": 0.3},
+            {"value": 0.1},
+        ]
+
+        with patch(
+            "timmy.sovereignty.session_report.get_sovereignty_store",
+            return_value=mock_store,
+        ):
+            with patch(
+                "timmy.sovereignty.session_report.GRADUATION_TARGETS",
+                {"cache_hit_rate": {"graduation": 0.9}},
+            ):
+                data = _gather_sovereignty_data()
+
+        delta = data["deltas"].get("cache_hit_rate")
+        assert delta is not None
+        assert delta["start"] == 0.1  # oldest in window
+        assert delta["end"] == 0.5    # most recent
+        assert data["previous_session"]["cache_hit_rate"] == 0.3
+
+    def test_single_data_point_no_delta(self):
+        mock_store = MagicMock()
+        mock_store.get_summary.return_value = {}
+        mock_store.get_latest.return_value = [{"value": 0.4}]
+
+        with patch(
+            "timmy.sovereignty.session_report.get_sovereignty_store",
+            return_value=mock_store,
+        ):
+            with patch(
+                "timmy.sovereignty.session_report.GRADUATION_TARGETS",
+                {"api_cost": {"graduation": 0.01}},
+            ):
+                data = _gather_sovereignty_data()
+
+        delta = data["deltas"]["api_cost"]
+        assert delta["start"] == 0.4
+        assert delta["end"] == 0.4
+        assert data["previous_session"]["api_cost"] is None
+
+
+# ---------------------------------------------------------------------------
+# generate_report (integration — smoke test)
+# ---------------------------------------------------------------------------
+
+
+class TestGenerateReport:
+    def _minimal_session_data(self):
+        return {
+            "user_messages": 3,
+            "timmy_messages": 3,
+            "tool_calls": 2,
+            "errors": 0,
+            "tool_call_breakdown": {"memory_search": 2},
+        }
+
+    def _minimal_sov_data(self):
+        return {
+            "metrics": {
+                "cache_hit_rate": {"current": 0.45, "phase": "week1"},
+                "api_cost": {"current": 0.12, "phase": "pre-start"},
+            },
+            "deltas": {
+                "cache_hit_rate": {"start": 0.40, "end": 0.45},
+                "api_cost": {"start": 0.10, "end": 0.12},
+            },
+            "previous_session": {
+                "cache_hit_rate": 0.40,
+                "api_cost": 0.10,
+            },
+        }
+
+    def test_smoke_produces_markdown(self):
+        with (
+            patch(
+                "timmy.sovereignty.session_report._gather_session_data",
+                return_value=self._minimal_session_data(),
+            ),
+            patch(
+                "timmy.sovereignty.session_report._gather_sovereignty_data",
+                return_value=self._minimal_sov_data(),
+            ),
+        ):
+            report = generate_report("test-session")
+
+        assert "# Sovereignty Session Report" in report
+        assert "test-session" in report
+        assert "## Session Activity" in report
+        assert "## Sovereignty Scorecard" in report
+        assert "## Cost Breakdown" in report
+        assert "## Trend vs Previous Session" in report
+
+    def test_report_contains_session_stats(self):
+        with (
+            patch(
+                "timmy.sovereignty.session_report._gather_session_data",
+                return_value=self._minimal_session_data(),
+            ),
+            patch(
+                "timmy.sovereignty.session_report._gather_sovereignty_data",
+                return_value=self._minimal_sov_data(),
+            ),
+        ):
+            report = generate_report()
+
+        assert "| User messages | 3 |" in report
+        assert "memory_search" in report
+
+    def test_report_no_previous_session(self):
+        sov = self._minimal_sov_data()
+        sov["previous_session"] = {"cache_hit_rate": None, "api_cost": None}
+
+        with (
+            patch(
+                "timmy.sovereignty.session_report._gather_session_data",
+                return_value=self._minimal_session_data(),
+            ),
+            patch(
+                "timmy.sovereignty.session_report._gather_sovereignty_data",
+                return_value=sov,
+            ),
+        ):
+            report = generate_report()
+
+        assert "No previous session data" in report
+
+
+# ---------------------------------------------------------------------------
+# commit_report
+# ---------------------------------------------------------------------------
+
+
+class TestCommitReport:
+    def test_returns_false_when_gitea_disabled(self):
+        with patch("timmy.sovereignty.session_report.settings") as mock_settings:
+            mock_settings.gitea_enabled = False
+            result = commit_report("# test", "dashboard")
+
+        assert result is False
+
+    def test_returns_false_when_no_token(self):
+        with patch("timmy.sovereignty.session_report.settings") as mock_settings:
+            mock_settings.gitea_enabled = True
+            mock_settings.gitea_token = ""
+            result = commit_report("# test", "dashboard")
+
+        assert result is False
+
+    def test_creates_file_via_put(self):
+        mock_response = MagicMock()
+        mock_response.status_code = 201
+        mock_response.raise_for_status.return_value = None
+
+        mock_check = MagicMock()
+        mock_check.status_code = 404  # file does not exist yet
+
+        mock_client = MagicMock()
+        mock_client.__enter__ = MagicMock(return_value=mock_client)
+        mock_client.__exit__ = MagicMock(return_value=False)
+        mock_client.get.return_value = mock_check
+        mock_client.put.return_value = mock_response
+
+        with (
+            patch("timmy.sovereignty.session_report.settings") as mock_settings,
+            patch("timmy.sovereignty.session_report.httpx.Client", return_value=mock_client),
+        ):
+            mock_settings.gitea_enabled = True
+            mock_settings.gitea_token = "fake-token"
+            mock_settings.gitea_url = "http://localhost:3000"
+            mock_settings.gitea_repo = "owner/repo"
+
+            result = commit_report("# report content", "dashboard")
+
+        assert result is True
+        mock_client.put.assert_called_once()
+        call_kwargs = mock_client.put.call_args
+        payload = call_kwargs.kwargs.get("json", call_kwargs.args[1] if len(call_kwargs.args) > 1 else {})
+        decoded = base64.b64decode(payload["content"]).decode()
+        assert "# report content" in decoded
+
+    def test_updates_existing_file_with_sha(self):
+        mock_check = MagicMock()
+        mock_check.status_code = 200
+        mock_check.json.return_value = {"sha": "abc123"}
+
+        mock_response = MagicMock()
+        mock_response.raise_for_status.return_value = None
+
+        mock_client = MagicMock()
+        mock_client.__enter__ = MagicMock(return_value=mock_client)
+        mock_client.__exit__ = MagicMock(return_value=False)
+        mock_client.get.return_value = mock_check
+        mock_client.put.return_value = mock_response
+
+        with (
+            patch("timmy.sovereignty.session_report.settings") as mock_settings,
+            patch("timmy.sovereignty.session_report.httpx.Client", return_value=mock_client),
+        ):
+            mock_settings.gitea_enabled = True
+            mock_settings.gitea_token = "fake-token"
+            mock_settings.gitea_url = "http://localhost:3000"
+            mock_settings.gitea_repo = "owner/repo"
+
+            result = commit_report("# updated", "dashboard")
+
+        assert result is True
+        payload = mock_client.put.call_args.kwargs.get("json", {})
+        assert payload.get("sha") == "abc123"
+
+    def test_returns_false_on_http_error(self):
+        import httpx
+
+        mock_check = MagicMock()
+        mock_check.status_code = 404
+
+        mock_client = MagicMock()
+        mock_client.__enter__ = MagicMock(return_value=mock_client)
+        mock_client.__exit__ = MagicMock(return_value=False)
+        mock_client.get.return_value = mock_check
+        mock_client.put.side_effect = httpx.HTTPStatusError(
+            "403", request=MagicMock(), response=MagicMock(status_code=403)
+        )
+
+        with (
+            patch("timmy.sovereignty.session_report.settings") as mock_settings,
+            patch("timmy.sovereignty.session_report.httpx.Client", return_value=mock_client),
+        ):
+            mock_settings.gitea_enabled = True
+            mock_settings.gitea_token = "fake-token"
+            mock_settings.gitea_url = "http://localhost:3000"
+            mock_settings.gitea_repo = "owner/repo"
+
+            result = commit_report("# test", "dashboard")
+
+        assert result is False
+
+
+# ---------------------------------------------------------------------------
+# generate_and_commit_report (async)
+# ---------------------------------------------------------------------------
+
+
+class TestGenerateAndCommitReport:
+    async def test_returns_true_on_success(self):
+        with (
+            patch(
+                "timmy.sovereignty.session_report.generate_report",
+                return_value="# mock report",
+            ),
+            patch(
+                "timmy.sovereignty.session_report.commit_report",
+                return_value=True,
+            ),
+        ):
+            result = await generate_and_commit_report("test")
+
+        assert result is True
+
+    async def test_returns_false_when_commit_fails(self):
+        with (
+            patch(
+                "timmy.sovereignty.session_report.generate_report",
+                return_value="# mock report",
+            ),
+            patch(
+                "timmy.sovereignty.session_report.commit_report",
+                return_value=False,
+            ),
+        ):
+            result = await generate_and_commit_report()
+
+        assert result is False
+
+    async def test_graceful_on_exception(self):
+        with patch(
+            "timmy.sovereignty.session_report.generate_report",
+            side_effect=RuntimeError("explode"),
+        ):
+            result = await generate_and_commit_report()
+
+        assert result is False