didericis
42004d37fd
Addresses the five review comments on PR #318: - Split PullRequest from Issue and add a dedicated read_pr method on Forge/ScopedForge/GiteaForge (a PR carries merge state an issue does not); is_pr_open now derives from read_pr. - Replace the JSON-file forge state with a thin swappable CRUD interface (ForgeStateStore) backed by SQLite (SqliteForgeStateStore) at ~/.bot-bottle/bot-bottle.db. - Remove the provenance footer (provenance.py + its test): a mutable, unsigned PR comment is not an audit record. - Reword the PRD: provenance is exposed via an API, not surfaced in the PR; document the Issue/PullRequest split and the SQLite store. pyright clean (whole repo), pylint 10/10, 38 forge/resume unit tests pass; no remaining refs to the removed provenance module or old JSON state API. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WL77TgFxKbs3cidGMG9dz7
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""Unit: SQLite forge state store (PRD forge-native-integration)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from bot_bottle.contrib.gitea.forge_state import (
|
|
STATUS_FROZEN,
|
|
STATUS_RUNNING,
|
|
ForgeState,
|
|
SqliteForgeStateStore,
|
|
)
|
|
|
|
|
|
def _state(**over: object) -> ForgeState:
|
|
base = ForgeState(
|
|
owner="didericis",
|
|
repo="bot-bottle",
|
|
issue_number=17,
|
|
slug="implementer-abc12",
|
|
agent_name="implementer",
|
|
bottle_names=["claude"],
|
|
backend_name="docker",
|
|
agent_git_user="didericis-claude",
|
|
pr_number=42,
|
|
status=STATUS_FROZEN,
|
|
last_checkin_at="2026-06-29T12:04:12-04:00",
|
|
)
|
|
return replace(base, **over)
|
|
|
|
|
|
class ForgeStateStoreTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
tmp = Path(self.enterContext(tempfile.TemporaryDirectory())) # pylint: disable=consider-using-with
|
|
self.store = SqliteForgeStateStore(tmp / "sub" / "bot-bottle.db")
|
|
|
|
def test_round_trip(self):
|
|
self.store.upsert(_state())
|
|
self.assertEqual(_state(), self.store.get("didericis", "bot-bottle", 17))
|
|
|
|
def test_missing_returns_none(self):
|
|
self.assertIsNone(self.store.get("nobody", "nope", 1))
|
|
|
|
def test_creates_db_parent_dirs(self):
|
|
# setUp pointed at a non-existent 'sub/' dir; init must create it.
|
|
self.assertIsNone(self.store.get("x", "y", 1)) # no raise
|
|
|
|
def test_upsert_replaces(self):
|
|
self.store.upsert(_state(status=STATUS_RUNNING))
|
|
self.store.upsert(_state(status=STATUS_FROZEN))
|
|
got = self.store.get("didericis", "bot-bottle", 17)
|
|
assert got is not None
|
|
self.assertEqual(STATUS_FROZEN, got.status)
|
|
# Still one row, not two.
|
|
self.assertEqual(1, len(self.store.all()))
|
|
|
|
def test_delete_is_idempotent(self):
|
|
self.store.upsert(_state())
|
|
self.store.delete("didericis", "bot-bottle", 17)
|
|
self.store.delete("didericis", "bot-bottle", 17) # no raise
|
|
self.assertIsNone(self.store.get("didericis", "bot-bottle", 17))
|
|
|
|
def test_all_lists_across_repos_sorted(self):
|
|
self.store.upsert(_state(issue_number=18, slug="other"))
|
|
self.store.upsert(_state(issue_number=17))
|
|
self.store.upsert(_state(owner="acme", repo="widget", issue_number=3))
|
|
states = self.store.all()
|
|
self.assertEqual(3, len(states))
|
|
self.assertEqual(
|
|
[("acme", 3), ("didericis", 17), ("didericis", 18)],
|
|
[(s.owner, s.issue_number) for s in states],
|
|
)
|
|
|
|
def test_all_empty(self):
|
|
self.assertEqual([], self.store.all())
|
|
|
|
def test_bottle_names_list_preserved(self):
|
|
self.store.upsert(_state(bottle_names=["claude", "dev"]))
|
|
got = self.store.get("didericis", "bot-bottle", 17)
|
|
assert got is not None
|
|
self.assertEqual(["claude", "dev"], got.bottle_names)
|
|
|
|
def test_pr_number_nullable(self):
|
|
self.store.upsert(_state(pr_number=None))
|
|
got = self.store.get("didericis", "bot-bottle", 17)
|
|
assert got is not None
|
|
self.assertIsNone(got.pr_number)
|
|
|
|
def test_persists_across_store_instances(self):
|
|
self.store.upsert(_state())
|
|
reopened = SqliteForgeStateStore(self.store._db_path) # pylint: disable=protected-access
|
|
self.assertEqual(_state(), reopened.get("didericis", "bot-bottle", 17))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|