Prepare rollback-safe legacy migration data contract (#7)

tests/unit/test_migration_maintenance.py

@@ -0,0 +1,45 @@
+from unittest.mock import Mock, patch
+
+import pytest
+from django.http import HttpResponse
+from django.test import RequestFactory
+
+from apps.core.middleware import MigrationMaintenanceMiddleware
+
+
+@pytest.mark.parametrize("method,path,allowed", [
+    ("GET", "/health/ready/", True),
+    ("HEAD", "/health/live/", True),
+    ("POST", "/health/ready/", False),
+    ("GET", "/health/ready", False),
+    ("GET", "/", False),
+    ("POST", "/accounts/login/", False),
+    ("OPTIONS", "/health/live/", False),
+])
+def test_held_candidate_only_allows_exact_safe_health(method, path, allowed):
+    downstream = Mock(return_value=HttpResponse("ok"))
+    with patch("apps.core.middleware.hold_active", return_value=True):
+        response = MigrationMaintenanceMiddleware(downstream)(
+            RequestFactory().generic(method, path)
+        )
+    assert response.status_code == (200 if allowed else 503)
+    assert downstream.called is allowed
+    if not allowed:
+        assert response["Cache-Control"] == "no-store"
+
+
+@pytest.mark.parametrize("failure", [PermissionError(), RuntimeError()])
+def test_marker_errors_block_ordinary_traffic(failure):
+    downstream = Mock()
+    with patch("apps.core.middleware.hold_active", side_effect=failure):
+        response = MigrationMaintenanceMiddleware(downstream)(RequestFactory().get("/"))
+    assert response.status_code == 503
+    downstream.assert_not_called()
+
+
+def test_normal_runtime_unchanged_and_middleware_is_first(settings):
+    assert settings.MIDDLEWARE[0] == "apps.core.middleware.MigrationMaintenanceMiddleware"
+    downstream = Mock(return_value=HttpResponse("ok"))
+    with patch("apps.core.middleware.hold_active", return_value=False):
+        response = MigrationMaintenanceMiddleware(downstream)(RequestFactory().post("/"))
+    assert response.status_code == 200

tests/unit/test_migration_worker_gate.py

@@ -0,0 +1,44 @@
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+spec = importlib.util.spec_from_file_location(
+    "migration_worker_gate",
+    Path(__file__).resolve().parents[2] / "scripts/migration_worker_gate.py",
+)
+gate = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(gate)
+
+
+@pytest.mark.parametrize("role", ["worker", "scheduler"])
+def test_hold_prevents_outbound_process_until_release(role):
+    states = iter([True, True, False])
+    events = []
+    gate.start(
+        role,
+        check=lambda: next(states),
+        sleep=lambda seconds: events.append("wait"),
+        execute=lambda executable, args: events.append(args),
+    )
+    assert events == ["wait", "wait", gate.COMMANDS[role]]
+
+
+def test_normal_startup_and_missing_hold(tmp_path):
+    assert not gate.hold_active(tmp_path / "absent")
+    events = []
+    gate.start("worker", check=lambda: False, execute=lambda executable, args: events.append(args))
+    assert events == [gate.COMMANDS["worker"]]
+
+
+def test_hold_exists_and_unsupported_marker_fails_closed(tmp_path):
+    marker = tmp_path / "hold"
+    marker.write_text("fixture-attempt")
+    assert gate.hold_active(marker)
+    with pytest.raises(RuntimeError):
+        gate.hold_active(tmp_path)
+
+
+def test_unknown_role_never_executes():
+    with pytest.raises(ValueError):
+        gate.start("shell", execute=lambda *args: pytest.fail("unreviewed command"))