#!/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()