feat(operator-gate): transport-agnostic idempotent Gitea mutation gate (#186)
Adds the canonical write path for all operator-originated Gitea mutations. Core: - gitea_gate.py with deterministic idempotency keys - Local ledger + Gitea-side probing for replay safety - Actions: create_issue, add_comment, close_issue, assign_issue, merge_pr Adapter: - nostur_adapter.py for Nostr DM ingress Tests: - idempotency, deduplication, and normalizer unit tests Docs: - README with architecture invariants and usage examples
This commit is contained in:
Timmy Foundation Ops
222
operator-gate/tests/test_gate_idempotency.py
Normal file
222
operator-gate/tests/test_gate_idempotency.py
Normal file
@@ -0,0 +1,222 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Idempotency and deduplication tests for the Operator Ingress Gate.
|
||||
|
||||
These tests verify the core contract:
|
||||
- Same command twice = one Gitea mutation + replay ACK
|
||||
- Duplicate title probe prevents double issue creation
|
||||
- Fingerprint probe prevents double comment
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from gitea_gate import Command, GiteaGate, _ledger_load, _LEDGER_PATH
|
||||
|
||||
|
||||
class TestIdempotency(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.TemporaryDirectory()
|
||||
self.ledger_override = os.path.join(self.tmpdir.name, "test_ledger.jsonl")
|
||||
self.gate = GiteaGate()
|
||||
|
||||
def tearDown(self):
|
||||
self.tmpdir.cleanup()
|
||||
|
||||
@patch("gitea_gate._LEDGER_PATH")
|
||||
def test_replay_returns_prior_ack(self, mock_path):
|
||||
"""Same command executed twice must return prior_execution=True on second call."""
|
||||
mock_path.__str__ = lambda self: self.ledger_override
|
||||
# We mock the API call to avoid network
|
||||
with patch("gitea_gate._api_request") as mock_api:
|
||||
mock_api.return_value = {"number": 999, "title": "Test Issue", "html_url": "http://test/999"}
|
||||
|
||||
cmd = Command(
|
||||
source="nostr:test",
|
||||
action="create_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"title": "Test Issue", "body": "body"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
|
||||
ack1 = self.gate.execute(cmd)
|
||||
self.assertTrue(ack1.success)
|
||||
self.assertFalse(ack1.prior_execution)
|
||||
|
||||
ack2 = self.gate.execute(cmd)
|
||||
self.assertTrue(ack2.success)
|
||||
self.assertTrue(ack2.prior_execution)
|
||||
self.assertEqual(ack2.gitea_number, 999)
|
||||
|
||||
def test_idempotency_key_determinism(self):
|
||||
"""Identical commands must produce identical keys."""
|
||||
cmd1 = Command(
|
||||
source="nostr:npub123",
|
||||
action="create_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"title": "X", "body": "Y"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
cmd2 = Command(
|
||||
source="nostr:npub123",
|
||||
action="create_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"title": "X", "body": "Y"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
self.assertEqual(cmd1.idempotency_key(), cmd2.idempotency_key())
|
||||
|
||||
def test_idempotency_key_uniqueness(self):
|
||||
"""Different payloads must produce different keys."""
|
||||
cmd1 = Command(
|
||||
source="nostr:npub123",
|
||||
action="create_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"title": "X", "body": "Y"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
cmd2 = Command(
|
||||
source="nostr:npub123",
|
||||
action="create_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"title": "X", "body": "Z"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
self.assertNotEqual(cmd1.idempotency_key(), cmd2.idempotency_key())
|
||||
|
||||
@patch("gitea_gate._api_get")
|
||||
@patch("gitea_gate._api_request")
|
||||
def test_duplicate_title_probe_prevents_double_create(self, mock_post, mock_get):
|
||||
"""If Gitea already has an issue with the same title, gate must return prior_execution."""
|
||||
mock_get.return_value = [
|
||||
{"number": 42, "title": "Probe Title", "html_url": "http://test/42"}
|
||||
]
|
||||
|
||||
cmd = Command(
|
||||
source="nostr:test",
|
||||
action="create_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"title": "Probe Title", "body": "body"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
|
||||
ack = self.gate.execute(cmd)
|
||||
self.assertTrue(ack.success)
|
||||
self.assertTrue(ack.prior_execution)
|
||||
self.assertEqual(ack.gitea_number, 42)
|
||||
mock_post.assert_not_called()
|
||||
|
||||
@patch("gitea_gate._api_get")
|
||||
@patch("gitea_gate._api_request")
|
||||
def test_duplicate_comment_probe_prevents_double_comment(self, mock_post, mock_get):
|
||||
"""If a comment with the same gate-key already exists, do not post again."""
|
||||
key = Command(
|
||||
source="nostr:test",
|
||||
action="add_comment",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"issue_num": 7, "body": "hello"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
).idempotency_key()
|
||||
|
||||
mock_get.return_value = [
|
||||
{"id": 101, "body": f"hello\n\n---\n*gate-key: `{key}`*"}
|
||||
]
|
||||
|
||||
cmd = Command(
|
||||
source="nostr:test",
|
||||
action="add_comment",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"issue_num": 7, "body": "hello"},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
|
||||
ack = self.gate.execute(cmd)
|
||||
self.assertTrue(ack.success)
|
||||
self.assertTrue(ack.prior_execution)
|
||||
mock_post.assert_not_called()
|
||||
|
||||
@patch("gitea_gate._api_get")
|
||||
@patch("gitea_gate._api_request")
|
||||
def test_already_closed_issue_returns_prior(self, mock_post, mock_get):
|
||||
mock_get.return_value = {"state": "closed", "number": 5, "html_url": "http://test/5"}
|
||||
|
||||
cmd = Command(
|
||||
source="nostr:test",
|
||||
action="close_issue",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"issue_num": 5},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
|
||||
ack = self.gate.execute(cmd)
|
||||
self.assertTrue(ack.success)
|
||||
self.assertTrue(ack.prior_execution)
|
||||
mock_post.assert_not_called()
|
||||
|
||||
@patch("gitea_gate._api_get")
|
||||
@patch("gitea_gate._api_request")
|
||||
def test_already_merged_pr_returns_prior(self, mock_post, mock_get):
|
||||
mock_get.return_value = {"merged": True, "number": 3, "html_url": "http://test/3"}
|
||||
|
||||
cmd = Command(
|
||||
source="nostr:test",
|
||||
action="merge_pr",
|
||||
repo="Timmy_Foundation/test",
|
||||
payload={"pr_num": 3},
|
||||
timestamp_utc="2026-04-06T14:00:00Z",
|
||||
)
|
||||
|
||||
ack = self.gate.execute(cmd)
|
||||
self.assertTrue(ack.success)
|
||||
self.assertTrue(ack.prior_execution)
|
||||
mock_post.assert_not_called()
|
||||
|
||||
|
||||
class TestNosturNormalizer(unittest.TestCase):
|
||||
def setUp(self):
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "adapters"))
|
||||
import nostur_adapter
|
||||
self.nostur = nostur_adapter
|
||||
|
||||
def test_status_command(self):
|
||||
cmd = self.nostur._normalize_dm("status")
|
||||
self.assertEqual(cmd["action"], "status")
|
||||
|
||||
def test_create_command(self):
|
||||
cmd = self.nostur._normalize_dm("create the-nexus Fix the bug\nDetails here")
|
||||
self.assertEqual(cmd["action"], "create_issue")
|
||||
self.assertEqual(cmd["repo"], "Timmy_Foundation/the-nexus")
|
||||
self.assertEqual(cmd["title"], "Fix the bug")
|
||||
self.assertEqual(cmd["body"], "Details here")
|
||||
|
||||
def test_comment_command(self):
|
||||
cmd = self.nostur._normalize_dm("comment the-nexus #42 This is the comment")
|
||||
self.assertEqual(cmd["action"], "add_comment")
|
||||
self.assertEqual(cmd["issue_num"], 42)
|
||||
self.assertEqual(cmd["body"], "This is the comment")
|
||||
|
||||
def test_close_command(self):
|
||||
cmd = self.nostur._normalize_dm("close the-nexus #7")
|
||||
self.assertEqual(cmd["action"], "close_issue")
|
||||
self.assertEqual(cmd["issue_num"], 7)
|
||||
|
||||
def test_assign_command(self):
|
||||
cmd = self.nostur._normalize_dm("assign the-nexus #7 allegro,ezra")
|
||||
self.assertEqual(cmd["action"], "assign_issue")
|
||||
self.assertEqual(cmd["assignees"], ["allegro", "ezra"])
|
||||
|
||||
def test_merge_command(self):
|
||||
cmd = self.nostur._normalize_dm("merge the-nexus #108")
|
||||
self.assertEqual(cmd["action"], "merge_pr")
|
||||
self.assertEqual(cmd["pr_num"], 108)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user