232 lines
8.0 KiB
Python
232 lines
8.0 KiB
Python
#!/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_load")
|
|
@patch("gitea_gate._ledger_append")
|
|
@patch("gitea_gate._api_request")
|
|
def test_replay_returns_prior_ack(self, mock_api, mock_append, mock_load):
|
|
"""Same command executed twice must return prior_execution=True on second call."""
|
|
mock_api.return_value = {"number": 999, "title": "Test Issue", "html_url": "http://test/999"}
|
|
ledger = {}
|
|
|
|
def load():
|
|
return ledger.copy()
|
|
|
|
def append(entry):
|
|
ledger[entry["idempotency_key"]] = entry
|
|
|
|
mock_load.side_effect = load
|
|
mock_append.side_effect = append
|
|
|
|
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()
|