didericis
f211ece6bf
CI runs `pyright .` over the whole repo including tests; the earlier run only checked the source paths. The test helpers used `**over` dict-splat into typed constructors, which pyright strict rejects. - forge_state: build a typed ForgeState base and dataclasses.replace(**over) - provenance: explicit typed keyword params instead of a **over dict - resume: _launch_kwargs returns dict[str, Any] (copy call_args.kwargs) - forge_base: assert PermissionError in __mro__ (avoids always-true issubclass) - client: annotate _resp body param; type: ignore the mock __enter__ lambda pyright . now 0 errors; 47 tests still pass; pylint 9.97/10. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WL77TgFxKbs3cidGMG9dz7
104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""Unit: forge state persistence (PRD forge-native-integration)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
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 ForgeStateTest(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)
|
|
|
|
def test_round_trip(self):
|
|
fs.write_forge_state(_state())
|
|
got = fs.read_forge_state("didericis", "bot-bottle", 17)
|
|
self.assertEqual(_state(), got)
|
|
|
|
def test_missing_returns_none(self):
|
|
self.assertIsNone(fs.read_forge_state("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_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)
|
|
assert got is not None
|
|
self.assertEqual(STATUS_FROZEN, got.status)
|
|
|
|
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))
|
|
|
|
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()
|
|
self.assertEqual(3, len(states))
|
|
self.assertEqual({17, 18, 3}, {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_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_pr_number_optional(self):
|
|
fs.write_forge_state(_state(pr_number=None))
|
|
got = fs.read_forge_state("didericis", "bot-bottle", 17)
|
|
assert got is not None
|
|
self.assertIsNone(got.pr_number)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|