refactor(forge): address PR #318 review — PR/Issue split, sqlite state, drop footer
lint / lint (push) Successful in 1m59s
test / unit (pull_request) Successful in 54s
test / integration (pull_request) Successful in 19s
test / coverage (pull_request) Successful in 1m4s

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
This commit is contained in:
2026-07-01 08:37:25 -04:00
parent f211ece6bf
commit 42004d37fd
9 changed files with 332 additions and 432 deletions
+45 -49
View File
@@ -1,4 +1,4 @@
"""Unit: forge state persistence (PRD forge-native-integration)."""
"""Unit: SQLite forge state store (PRD forge-native-integration)."""
from __future__ import annotations
@@ -6,13 +6,12 @@ import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
from unittest.mock import patch
from bot_bottle.contrib.gitea import forge_state as fs
from bot_bottle.contrib.gitea.forge_state import (
STATUS_FROZEN,
STATUS_RUNNING,
ForgeState,
SqliteForgeStateStore,
)
@@ -33,71 +32,68 @@ def _state(**over: object) -> ForgeState:
return replace(base, **over)
class ForgeStateTest(unittest.TestCase):
class ForgeStateStoreTest(unittest.TestCase):
def setUp(self) -> None:
# enterContext handles cleanup; pylint doesn't recognize it as CM-aware.
root = Path(self.enterContext( # pylint: disable=consider-using-with
tempfile.TemporaryDirectory()))
patcher = patch.object(fs, "bot_bottle_root", return_value=root)
patcher.start()
self.addCleanup(patcher.stop)
tmp = Path(self.enterContext(tempfile.TemporaryDirectory())) # pylint: disable=consider-using-with
self.store = SqliteForgeStateStore(tmp / "sub" / "bot-bottle.db")
def test_round_trip(self):
fs.write_forge_state(_state())
got = fs.read_forge_state("didericis", "bot-bottle", 17)
self.assertEqual(_state(), got)
self.store.upsert(_state())
self.assertEqual(_state(), self.store.get("didericis", "bot-bottle", 17))
def test_missing_returns_none(self):
self.assertIsNone(fs.read_forge_state("nobody", "nope", 1))
self.assertIsNone(self.store.get("nobody", "nope", 1))
def test_path_layout(self):
path = fs.forge_state_path("didericis", "bot-bottle", 17)
self.assertTrue(str(path).endswith("forge/didericis/bot-bottle/issue-17.json"))
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_write_is_atomic_no_tmp_left(self):
fs.write_forge_state(_state())
path = fs.forge_state_path("didericis", "bot-bottle", 17)
self.assertFalse(path.with_suffix(".json.tmp").exists())
self.assertTrue(path.exists())
def test_update_overwrites(self):
fs.write_forge_state(_state(status=STATUS_RUNNING))
fs.write_forge_state(_state(status=STATUS_FROZEN))
got = fs.read_forge_state("didericis", "bot-bottle", 17)
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):
fs.write_forge_state(_state())
fs.delete_forge_state("didericis", "bot-bottle", 17)
fs.delete_forge_state("didericis", "bot-bottle", 17) # no raise
self.assertIsNone(fs.read_forge_state("didericis", "bot-bottle", 17))
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_forge_states_lists_across_repos(self):
fs.write_forge_state(_state(issue_number=17))
fs.write_forge_state(_state(issue_number=18, slug="other"))
fs.write_forge_state(_state(owner="acme", repo="widget", issue_number=3))
states = fs.all_forge_states()
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({17, 18, 3}, {s.issue_number for s in states})
self.assertEqual(
[("acme", 3), ("didericis", 17), ("didericis", 18)],
[(s.owner, s.issue_number) for s in states],
)
def test_all_forge_states_empty_when_no_dir(self):
self.assertEqual([], fs.all_forge_states())
def test_all_empty(self):
self.assertEqual([], self.store.all())
def test_from_dict_ignores_unknown_keys(self):
st = ForgeState.from_dict({
"owner": "o", "repo": "r", "issue_number": 1, "slug": "s",
"agent_name": "a", "future_field": "ignored",
})
self.assertEqual("o", st.owner)
self.assertIsNone(st.pr_number)
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_optional(self):
fs.write_forge_state(_state(pr_number=None))
got = fs.read_forge_state("didericis", "bot-bottle", 17)
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()