import hashlib import json import os import sys import tempfile import unittest from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCRIPT_DIR = ROOT / "scripts" / "salesforce_migration" sys.path.insert(0, str(SCRIPT_DIR)) sys.path.insert(0, str(ROOT / "scripts")) import build_incremental_sync_evidence as evidence_builder # noqa: E402 import notify_feishu_lead_incremental_sync_failure as failure_notifier # noqa: E402 import recover_hourly_delta_replay_entry as replay_recovery # noqa: E402 import replay_lead_assignment_from_export as replay_export # noqa: E402 import run_hourly_delta_catchup as catchup # noqa: E402 import run_hourly_delta_replay_queue as replay_queue # noqa: E402 from incremental_sync_state import ( # noqa: E402 persist_replay_intent, sanitize_error_category_counts, ) class ReplayProgressTests(unittest.TestCase): def test_new_replay_intent_starts_with_no_no_progress_failures(self): with tempfile.TemporaryDirectory() as temp_dir: queue_file = Path(temp_dir) / "queue.json" entry = persist_replay_intent( queue_file, entry_id="2026-08-18T00:30:00Z__2026-08-18T01:00:00Z", window_start="2026-08-18T00:30:00Z", window_end="2026-08-18T01:00:00Z", description_field="Description", lead_ids=["lead-1"], updated_at="2026-08-18T02:00:00Z", ) self.assertEqual(0, entry["consecutive_no_progress_count"]) def test_progress_resets_consecutive_no_progress_before_dead_lettering(self): entry = { "attempt_count": 5, "consecutive_no_progress_count": 4, "lead_ids": ["lead-1", "lead-2", "lead-3"], "completed_lead_ids": [], "all_candidate_ids": ["lead-1", "lead-2", "lead-3"], } replay_queue.record_replay_result( entry, attempted_lead_ids=["lead-1", "lead-2", "lead-3"], success_ids={"lead-1", "lead-2"}, failure_count=1, missing_target_count=0, max_consecutive_no_progress=5, error_category_counts={"http_429": 1, "raw secret response": 1}, ) self.assertEqual("failed", entry["status"]) self.assertEqual(0, entry["consecutive_no_progress_count"]) self.assertEqual(["lead-3"], entry["lead_ids"]) self.assertEqual(1, entry["candidate_count"]) self.assertEqual(5, entry["attempt_count"]) self.assertEqual({"http_429": 1}, entry["last_error_category_counts"]) for expected_count in range(1, 5): replay_queue.record_replay_result( entry, attempted_lead_ids=["lead-3"], success_ids=set(), failure_count=1, missing_target_count=0, max_consecutive_no_progress=5, ) self.assertEqual("failed", entry["status"]) self.assertEqual( expected_count, entry["consecutive_no_progress_count"], ) replay_queue.record_replay_result( entry, attempted_lead_ids=["lead-3"], success_ids=set(), failure_count=1, missing_target_count=0, max_consecutive_no_progress=5, ) self.assertEqual("dead_letter", entry["status"]) self.assertEqual(5, entry["consecutive_no_progress_count"]) class ReplayErrorCategoryTests(unittest.TestCase): def test_error_category_counts_use_a_fixed_non_negative_allowlist(self): self.assertEqual( {"http_429": 3, "timeout": 1}, sanitize_error_category_counts( { "http_429": "3", "timeout": 1, "raw secret response": 99, "http_5xx": -2, } ), ) def test_failure_categories_are_counted_without_raw_error_text(self): failures = [ { "id": "lead-1", "error": "PATCH https://example.invalid failed: status=429 body=secret", }, {"id": "lead-2", "error": "HTTP Error 503: unavailable"}, {"id": "lead-3", "error": "The read operation timed out"}, {"id": "lead-4", "error": "Connection reset by peer"}, {"id": "lead-5", "error": "PATCH failed: status=400 body=private"}, {"id": "lead-6", "error": "unexpected response"}, ] counts = replay_export.build_error_category_counts( failures, missing_target_count=2, ) self.assertEqual( { "http_429": 1, "http_5xx": 1, "timeout": 1, "network": 1, "http_4xx": 1, "other": 1, "missing_target": 2, }, counts, ) self.assertNotIn("secret", str(counts)) self.assertNotIn("lead-1", str(counts)) class CatchupReplayFailureTests(unittest.TestCase): def test_failure_window_carries_redacted_replay_progress(self): error = ValueError("replay failed") error.catchup_failure_stage = "replay_execution" error.catchup_window_summary_path = "/tmp/window.json" error.replay_status = "failed" error.replay_attempt_count = 7 error.replay_consecutive_no_progress_count = 2 error.replay_remaining_candidate_count = 3 error.replay_error_category_counts = { "http_5xx": 2, "private error": 1, } result = catchup.build_failure_window( error, window_start=datetime(2026, 8, 18, 0, 30, tzinfo=timezone.utc), window_end=datetime(2026, 8, 18, 1, 0, tzinfo=timezone.utc), run_label_prefix="lead-campaign-incremental", ) self.assertEqual("failed", result["replay_status"]) self.assertEqual(7, result["replay_attempt_count"]) self.assertEqual(2, result["replay_consecutive_no_progress_count"]) self.assertEqual(3, result["replay_remaining_candidate_count"]) self.assertEqual({"http_5xx": 2}, result["replay_error_category_counts"]) class ReplayEvidenceTests(unittest.TestCase): def test_failure_notification_summarizes_replay_progress_and_categories(self): summary = failure_notifier.summarize_evidence( { "catchup_summary_found": True, "run": {"mode": "write"}, "catchup": {}, "durable_state": {"state_unchanged": True}, "windows": [ { "status": "failed", "window_start": "2026-08-18T00:30:00Z", "window_end": "2026-08-18T01:00:00Z", "failure_stage": "replay_execution", "error_type": "ValueError", "replay_status": "dead_letter", "replay_attempt_count": 8, "replay_consecutive_no_progress_count": 5, "replay_remaining_candidate_count": 3, "replay_error_category_counts": { "http_429": 2, "timeout": 1, }, } ], } ) self.assertEqual( "dead_letter; attempts=8; consecutive_no_progress=5; remaining=3", summary["Replay"], ) self.assertEqual("http_429=2, timeout=1", summary["Replay errors"]) def test_evidence_preserves_redacted_replay_failure_counts(self): with tempfile.TemporaryDirectory() as temp_dir: summary_path = Path(temp_dir) / "catchup.json" summary_path.write_text( """{ "effective_start": "2026-08-18T00:30:00Z", "effective_until": "2026-08-18T01:00:00Z", "planned_window_count": 1, "windows": [{ "window_start": "2026-08-18T00:30:00Z", "window_end": "2026-08-18T01:00:00Z", "status": "failed", "failure_stage": "replay_execution", "error_type": "ValueError", "replay_status": "dead_letter", "replay_attempt_count": 8, "replay_consecutive_no_progress_count": 5, "replay_remaining_candidate_count": 3, "replay_error_category_counts": { "http_429": 2, "private response": 99 } }] }\n""", encoding="utf-8", ) evidence = evidence_builder.build_evidence( catchup_summary_path=summary_path, sync_outcome="failure", mode="write", state_before_sha256="missing", state_after_sha256="missing", replay_before_sha256="missing", replay_after_sha256="missing", head_sha="a" * 40, run_id="123", run_attempt="1", ) window = evidence["windows"][0] self.assertEqual(5, window["replay_consecutive_no_progress_count"]) self.assertEqual(3, window["replay_remaining_candidate_count"]) self.assertEqual( {"http_429": 2}, window["replay_error_category_counts"], ) self.assertNotIn("private response", str(evidence)) class ReplayRecoveryTests(unittest.TestCase): ENTRY_ID = "2026-08-18T00:30:00Z__2026-08-18T01:00:00Z" def _write_fixture(self, root: Path, *, cursor: str) -> tuple[Path, Path, str]: queue_file = root / "queue.json" state_file = root / "state.json" queue_file.write_text( json.dumps( { "version": 1, "entries": [ { "id": self.ENTRY_ID, "window_start": "2026-08-18T00:30:00Z", "window_end": "2026-08-18T01:00:00Z", "status": "dead_letter", "attempt_count": 8, "consecutive_no_progress_count": 5, "lead_ids": ["lead-1", "lead-2"], "candidate_count": 2, "all_candidate_ids": ["lead-1", "lead-2", "lead-3"], "completed_lead_ids": ["lead-3"], } ], }, indent=2, ) + "\n", encoding="utf-8", ) os.chmod(queue_file, 0o640) state_file.write_text( json.dumps({"version": 1, "next_window_start": cursor}) + "\n", encoding="utf-8", ) queue_sha = hashlib.sha256(queue_file.read_bytes()).hexdigest() return queue_file, state_file, queue_sha def test_exact_cursor_compare_and_swap_recovers_dead_letter_with_backup(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) queue_file, state_file, queue_sha = self._write_fixture( root, cursor="2026-08-18T00:30:00Z", ) result = replay_recovery.recover_replay_entry( queue_file=queue_file, state_file=state_file, entry_id=self.ENTRY_ID, expected_queue_sha256=queue_sha, recovered_at="2026-08-18T05:00:00Z", actor="github-actions", run_id="321", backup_suffix="run-321", ) payload = json.loads(queue_file.read_text(encoding="utf-8")) entry = payload["entries"][0] backup_path = Path(result["backup_path"]) self.assertEqual("failed", entry["status"]) self.assertEqual(8, entry["attempt_count"]) self.assertEqual(0, entry["consecutive_no_progress_count"]) self.assertEqual(["lead-1", "lead-2"], entry["lead_ids"]) self.assertEqual( queue_sha, hashlib.sha256(backup_path.read_bytes()).hexdigest() ) self.assertEqual(0o640, queue_file.stat().st_mode & 0o777) self.assertEqual(0o640, backup_path.stat().st_mode & 0o777) self.assertEqual(queue_sha, entry["last_recovery"]["preimage_sha256"]) self.assertEqual(1, entry["recovery_count"]) def test_recovery_rejects_entry_that_is_not_at_current_cursor(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) queue_file, state_file, queue_sha = self._write_fixture( root, cursor="2026-08-18T01:00:00Z", ) before = queue_file.read_bytes() with self.assertRaisesRegex(ValueError, "current cursor"): replay_recovery.recover_replay_entry( queue_file=queue_file, state_file=state_file, entry_id=self.ENTRY_ID, expected_queue_sha256=queue_sha, recovered_at="2026-08-18T05:00:00Z", actor="github-actions", run_id="321", backup_suffix="run-321", ) self.assertEqual(before, queue_file.read_bytes()) self.assertEqual([], list(root.glob("queue.json.bak-*"))) def test_recovery_rejects_stale_queue_sha_without_writing_backup(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) queue_file, state_file, _ = self._write_fixture( root, cursor="2026-08-18T00:30:00Z", ) before = queue_file.read_bytes() with self.assertRaisesRegex(ValueError, "compare-and-swap failed"): replay_recovery.recover_replay_entry( queue_file=queue_file, state_file=state_file, entry_id=self.ENTRY_ID, expected_queue_sha256="0" * 64, recovered_at="2026-08-18T05:00:00Z", actor="github-actions", run_id="321", backup_suffix="run-321", ) self.assertEqual(before, queue_file.read_bytes()) self.assertEqual([], list(root.glob("queue.json.bak-*"))) def test_recovery_rejects_symlinked_queue_file(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) queue_file, state_file, queue_sha = self._write_fixture( root, cursor="2026-08-18T00:30:00Z", ) queue_link = root / "queue-link.json" queue_link.symlink_to(queue_file) before = queue_file.read_bytes() with self.assertRaisesRegex(ValueError, "must not be a symlink"): replay_recovery.recover_replay_entry( queue_file=queue_link, state_file=state_file, entry_id=self.ENTRY_ID, expected_queue_sha256=queue_sha, recovered_at="2026-08-18T05:00:00Z", actor="github-actions", run_id="321", backup_suffix="run-321", ) self.assertEqual(before, queue_file.read_bytes()) if __name__ == "__main__": unittest.main()