Merge pull request #460 from AIOSAI/system/devpulse-fix-python-310-mockpatch-ensure-detector-module-im

feat(system): fix: Python 3.10 mock.patch — ensure detector module imported before patching in rollover tests

src/aipass/api/tests/test_integrations.py

@@ -171,6 +171,29 @@ class TestFetchContracts:
         assert result["contracts"] == ["alpha", "beta"]
         assert result["count"] == 2
 
+    def test_get_contracts_returns_expected_shape(self):
+        """get_contracts() result dict has contracts, count, and success keys."""
+        bridge.register("shaped", lambda: None)
+        result = fetch_contracts()
+        assert isinstance(result["contracts"], list)
+        assert isinstance(result["count"], int)
+        assert isinstance(result["success"], bool)
+        assert result["count"] == len(result["contracts"])
+
+    def test_get_contracts_handles_logging_failure(self):
+        """get_contracts() returns graceful failure when json_handler raises."""
+        from unittest.mock import patch
+
+        bridge.register("test_logging", lambda: None)
+        with patch(
+            "aipass.api.apps.handlers.integrations.list.json_handler.log_operation",
+            side_effect=RuntimeError("log broke"),
+        ):
+            result = fetch_contracts()
+        assert result["success"] is False
+        assert result["contracts"] == []
+        assert result["count"] == 0
+
 
 # ---------------------------------------------------------------------------
 # TestCallContract

src/aipass/flow/tests/test_lock_ops.py

@@ -0,0 +1,207 @@
+# =================== AIPass ====================
+# Name: test_lock_ops.py
+# Description: Tests for lock_ops handler — atomic lock file management
+# Version: 1.0.0
+# Created: 2026-04-26
+# Modified: 2026-04-26
+# =============================================
+
+"""Tests for lock_ops handler — atomic lock file management."""
+
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+
+# ─── Patch targets ───────────────────────────────────────
+_MOD = "aipass.flow.apps.handlers.runner.lock_ops"
+
+
+def _import_lock_ops():
+    """Import lock_ops module and return it."""
+    import aipass.flow.apps.handlers.runner.lock_ops as mod
+
+    return mod
+
+
+# ═══════════════════════════════════════════════════════════
+# 1. try_create_lock
+# ═══════════════════════════════════════════════════════════
+
+
+class TestTryCreateLock:
+    """Tests for try_create_lock — atomic O_CREAT|O_EXCL lock creation."""
+
+    def test_creates_lock_file_successfully(self, tmp_path):
+        """Should create lock file with current PID and return True."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        result = mod.try_create_lock(lock)
+        assert result is True
+        assert lock.exists()
+        assert lock.read_text(encoding="utf-8") == str(os.getpid())
+
+    def test_returns_false_if_lock_exists(self, tmp_path):
+        """Should return False when lock file already exists."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("12345", encoding="utf-8")
+        result = mod.try_create_lock(lock)
+        assert result is False
+
+    def test_does_not_overwrite_existing_lock(self, tmp_path):
+        """Existing lock content should be preserved on failure."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("99999", encoding="utf-8")
+        mod.try_create_lock(lock)
+        assert lock.read_text(encoding="utf-8") == "99999"
+
+
+# ═══════════════════════════════════════════════════════════
+# 2. is_lock_stale
+# ═══════════════════════════════════════════════════════════
+
+
+class TestIsLockStale:
+    """Tests for is_lock_stale — dead process detection."""
+
+    def test_lock_with_current_pid_is_not_stale(self, tmp_path):
+        """Lock file holding current PID should not be considered stale."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text(str(os.getpid()), encoding="utf-8")
+        result = mod.is_lock_stale(lock)
+        assert result is False
+
+    def test_lock_with_dead_pid_is_stale(self, tmp_path):
+        """Lock file holding a non-existent PID should be stale."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("999999999", encoding="utf-8")
+        with patch(f"{_MOD}.os.kill", side_effect=ProcessLookupError):
+            result = mod.is_lock_stale(lock)
+        assert result is True
+
+    def test_lock_with_invalid_content_is_stale(self, tmp_path):
+        """Lock file with non-integer content should be stale."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("not-a-pid", encoding="utf-8")
+        result = mod.is_lock_stale(lock)
+        assert result is True
+
+    def test_lock_with_empty_content_is_stale(self, tmp_path):
+        """Lock file with empty content should be stale."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("", encoding="utf-8")
+        result = mod.is_lock_stale(lock)
+        assert result is True
+
+    def test_permission_error_treated_as_stale(self, tmp_path):
+        """PermissionError from os.kill should treat lock as stale."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("1", encoding="utf-8")
+        with patch(f"{_MOD}.os.kill", side_effect=PermissionError):
+            result = mod.is_lock_stale(lock)
+        assert result is True
+
+
+# ═══════════════════════════════════════════════════════════
+# 3. acquire_lock
+# ═══════════════════════════════════════════════════════════
+
+
+class TestAcquireLock:
+    """Tests for acquire_lock — full lock acquisition with stale recovery."""
+
+    def test_acquires_fresh_lock(self, tmp_path):
+        """Should acquire lock when no lock file exists."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        result = mod.acquire_lock(lock)
+        assert result is True
+        assert lock.exists()
+
+    def test_fails_when_another_process_holds_lock(self, tmp_path):
+        """Should return False when lock held by a live process."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text(str(os.getpid()), encoding="utf-8")
+        result = mod.acquire_lock(lock)
+        assert result is False
+
+    def test_recovers_stale_lock(self, tmp_path):
+        """Should recover a stale lock (dead PID) and acquire it."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("999999999", encoding="utf-8")
+        with patch(f"{_MOD}.os.kill", side_effect=ProcessLookupError):
+            result = mod.acquire_lock(lock)
+        assert result is True
+        assert lock.read_text(encoding="utf-8") == str(os.getpid())
+
+    def test_fails_when_stale_lock_unlink_fails(self, tmp_path):
+        """Should return False when stale lock can't be removed."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("999999999", encoding="utf-8")
+        with (
+            patch(f"{_MOD}.os.kill", side_effect=ProcessLookupError),
+            patch.object(Path, "unlink", side_effect=OSError("permission denied")),
+        ):
+            result = mod.acquire_lock(lock)
+        assert result is False
+
+    def test_logs_json_operation_on_fresh_acquire(self, tmp_path, mock_json_handler):
+        """Should log lock_acquired via json_handler on fresh lock."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        mod.acquire_lock(lock)
+        mock_json_handler.assert_called()
+        call_args = mock_json_handler.call_args
+        assert call_args[0][0] == "lock_acquired"
+        assert call_args[0][1]["lock_file"] == str(lock)
+
+    def test_logs_stale_recovery_on_stale_acquire(self, tmp_path, mock_json_handler):
+        """Should log stale_recovery=True when recovering stale lock."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("999999999", encoding="utf-8")
+        with patch(f"{_MOD}.os.kill", side_effect=ProcessLookupError):
+            mod.acquire_lock(lock)
+        call_args = mock_json_handler.call_args
+        assert call_args[0][1]["stale_recovery"] is True
+
+
+# ═══════════════════════════════════════════════════════════
+# 4. release_lock
+# ═══════════════════════════════════════════════════════════
+
+
+class TestReleaseLock:
+    """Tests for release_lock — lock file cleanup."""
+
+    def test_removes_existing_lock(self, tmp_path):
+        """Should remove the lock file."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text(str(os.getpid()), encoding="utf-8")
+        mod.release_lock(lock)
+        assert not lock.exists()
+
+    def test_no_error_when_lock_missing(self, tmp_path):
+        """Should handle missing lock file gracefully (missing_ok=True)."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".nonexistent.lock"
+        mod.release_lock(lock)
+
+    def test_logs_warning_on_os_error(self, tmp_path, mock_logger):
+        """Should log warning when lock removal fails."""
+        mod = _import_lock_ops()
+        lock = tmp_path / ".test.lock"
+        lock.write_text("12345", encoding="utf-8")
+        with patch.object(Path, "unlink", side_effect=OSError("disk error")):
+            mod.release_lock(lock)

src/aipass/spawn/templates/builder/.spawn/.template_registry.json

@@ -1,7 +1,7 @@
 {
   "metadata": {
     "version": "1.0.0",
-    "last_updated": "2026-04-22",
+    "last_updated": "2026-04-26",
     "description": "Template file tracking registry for ID-based updates"
   },
   "files": {
@@ -128,7 +128,7 @@
     "f020": {
       "path": "apps/__init__.py",
       "name": "__init__.py",
-      "content_hash": "41b011f487af",
+      "content_hash": "0ef6c27137dc",
       "has_branch_placeholder": false
     },
     "f021": {
@@ -155,7 +155,7 @@
       "content_hash": "a4cf0a8e3b4f",
       "has_branch_placeholder": false
     },
-    "f015": {
+    "f026": {
       "path": "apps/modules/__init__.py",
       "name": "__init__.py",
       "content_hash": "e3b0c44298fc",
@@ -263,7 +263,7 @@
       "content_hash": "28e9ae373563",
       "has_branch_placeholder": false
     },
-    "f026": {
+    "f015": {
       "path": "apps/plugins/__init__.py",
       "name": "__init__.py",
       "content_hash": "e3b0c44298fc",

src/aipass/spawn/tests/test_regenerate_registry_ops.py

@@ -17,6 +17,7 @@ from aipass.spawn.apps.handlers.regenerate_registry_ops import (
     _scan_template_directory,
     _next_id,
 )
+from aipass.spawn.apps.modules.regenerate_registry import handle_regenerate_registry
 
 
 # =============================================================================
@@ -555,3 +556,31 @@ class TestScanTemplateDirectoryOrdering:
         # Directories should also be in sorted order
         dir_paths = [entry["path"] for entry in directories.values()]
         assert dir_paths == sorted(dir_paths)
+
+
+class TestHandleRegenerateRegistry:
+    """Tests for handle_regenerate_registry() CLI handler."""
+
+    def test_default_regenerates_builder(self):
+        result = handle_regenerate_registry([])
+        assert result == 0
+
+    def test_explicit_builder(self):
+        result = handle_regenerate_registry(["builder"])
+        assert result == 0
+
+    def test_explicit_birthright(self):
+        result = handle_regenerate_registry(["birthright"])
+        assert result == 0
+
+    def test_all_flag(self):
+        result = handle_regenerate_registry(["--all"])
+        assert result == 0
+
+    def test_help_flag(self):
+        result = handle_regenerate_registry(["--help"])
+        assert result == 0
+
+    def test_unknown_class_returns_error(self):
+        result = handle_regenerate_registry(["nonexistent_class"])
+        assert result == 1

src/aipass/spawn/tests/test_spawn.py

@@ -15,8 +15,12 @@ from pathlib import Path
 
 from aipass.spawn import spawn_agent
 from aipass.spawn.apps.handlers.metadata import get_branch_name, normalize_branch_name, detect_profile
-from aipass.spawn.apps.handlers.placeholders import build_replacements_dict, validate_no_placeholders
-from aipass.spawn.apps.handlers.registry import load_registry, add_to_registry
+from aipass.spawn.apps.handlers.placeholders import (
+    build_replacements_dict,
+    replace_placeholders,
+    validate_no_placeholders,
+)
+from aipass.spawn.apps.handlers.registry import load_registry, add_to_registry, save_registry, get_next_citizen_number
 
 
 @pytest.fixture
@@ -129,3 +133,90 @@ class TestSpawnAgent:
         data = json.loads(reg_file.read_text())
         assert data["metadata"]["generated"] is True
         assert len(data["files"]) > 0
+
+
+class TestReplacePlaceholders:
+    """Tests for replace_placeholders()."""
+
+    def test_basic_replacement(self):
+        content = "Hello {{NAME}}, welcome to {{PROJECT}}."
+        result = replace_placeholders(content, {"NAME": "Alice", "PROJECT": "AIPass"})
+        assert result == "Hello Alice, welcome to AIPass."
+
+    def test_no_placeholders(self):
+        content = "No placeholders here."
+        result = replace_placeholders(content, {"FOO": "bar"})
+        assert result == "No placeholders here."
+
+    def test_multiple_occurrences(self):
+        content = "{{X}} and {{X}} again"
+        result = replace_placeholders(content, {"X": "val"})
+        assert result == "val and val again"
+
+    def test_empty_replacements(self):
+        content = "{{STAYS}} intact"
+        result = replace_placeholders(content, {})
+        assert result == "{{STAYS}} intact"
+
+    def test_value_coercion_to_string(self):
+        result = replace_placeholders("number={{N}}", {"N": 42})
+        assert result == "number=42"
+
+
+class TestSaveRegistry:
+    """Tests for save_registry()."""
+
+    def test_saves_and_sorts_branches(self, tmp_path):
+        reg_path = tmp_path / "TEST_REGISTRY.json"
+        data = {
+            "metadata": {"version": "1.0.0", "total_branches": 2},
+            "branches": [
+                {"name": "ZEBRA", "path": "zebra"},
+                {"name": "ALPHA", "path": "alpha"},
+            ],
+        }
+        save_registry(reg_path, data)
+        saved = json.loads(reg_path.read_text())
+        assert saved["branches"][0]["name"] == "ALPHA"
+        assert saved["branches"][1]["name"] == "ZEBRA"
+
+    def test_updates_timestamp(self, tmp_path):
+        reg_path = tmp_path / "TEST_REGISTRY.json"
+        data = {
+            "metadata": {"version": "1.0.0", "last_updated": "2020-01-01", "total_branches": 0},
+            "branches": [],
+        }
+        save_registry(reg_path, data)
+        saved = json.loads(reg_path.read_text())
+        assert saved["metadata"]["last_updated"] != "2020-01-01"
+
+    def test_handles_dict_branches(self, tmp_path):
+        reg_path = tmp_path / "TEST_REGISTRY.json"
+        data = {
+            "metadata": {"version": "1.0.0", "total_branches": 1},
+            "branches": {"MY_AGENT": {"name": "MY_AGENT", "path": "my_agent"}},
+        }
+        save_registry(reg_path, data)
+        saved = json.loads(reg_path.read_text())
+        assert isinstance(saved["branches"], list)
+        assert saved["branches"][0]["name"] == "MY_AGENT"
+
+
+class TestGetNextCitizenNumber:
+    """Tests for get_next_citizen_number()."""
+
+    def test_empty_registry(self, tmp_path):
+        reg_path = tmp_path / "TEST_REGISTRY.json"
+        reg_path.write_text('{"metadata":{"version":"1.0.0","total_branches":0},"branches":[]}')
+        assert get_next_citizen_number(reg_path) == 1
+
+    def test_with_existing_branches(self, tmp_path):
+        reg_path = tmp_path / "TEST_REGISTRY.json"
+        reg_path.write_text(
+            '{"metadata":{"version":"1.0.0","total_branches":2},"branches":[{"name":"A"},{"name":"B"}]}'
+        )
+        assert get_next_citizen_number(reg_path) == 3
+
+    def test_missing_registry(self, tmp_path):
+        reg_path = tmp_path / "NONEXISTENT_REGISTRY.json"
+        assert get_next_citizen_number(reg_path) == 1