346 lines
11 KiB
Python
346 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import http.client
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
BROKER_DIR = REPO_ROOT / "ops" / "gitea-deploy-broker"
|
|
|
|
|
|
def _load_module(name: str, path: Path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.fixture
|
|
def storage_effects():
|
|
_load_module("repopulse_adoption_store", BROKER_DIR / "repopulse_adoption_store.py")
|
|
return _load_module(
|
|
"repopulse_adoption_storage_effects_test",
|
|
BROKER_DIR / "repopulse_adoption_storage_effects.py",
|
|
)
|
|
|
|
|
|
def _run_git(*args: str, cwd: Path | None = None) -> str:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=cwd,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def _bare_repo_with_helper(tmp_path: Path, helper: bytes) -> tuple[Path, str]:
|
|
worktree = tmp_path / "worktree"
|
|
bare_repo = tmp_path / "repositories" / "jens" / "repopulse.git"
|
|
worktree.mkdir()
|
|
bare_repo.parent.mkdir(parents=True)
|
|
_run_git("init", "--bare", str(bare_repo))
|
|
_run_git("init", "--initial-branch=main", str(worktree))
|
|
_run_git("config", "user.name", "Storage effects test", cwd=worktree)
|
|
_run_git("config", "user.email", "storage-effects@example.invalid", cwd=worktree)
|
|
helper_path = worktree / "scripts" / "repopulse_db_migration.py"
|
|
helper_path.parent.mkdir()
|
|
helper_path.write_bytes(helper)
|
|
_run_git("add", "scripts/repopulse_db_migration.py", cwd=worktree)
|
|
_run_git("commit", "-m", "Add reviewed migration helper", cwd=worktree)
|
|
_run_git("remote", "add", "origin", str(bare_repo), cwd=worktree)
|
|
_run_git("push", "origin", "main", cwd=worktree)
|
|
return bare_repo, _run_git("rev-parse", "HEAD", cwd=worktree)
|
|
|
|
|
|
def test_maintenance_http_error_is_sanitized(storage_effects):
|
|
with pytest.raises(storage_effects.StorageEffectsError) as error:
|
|
storage_effects._maintenance_is_drained(
|
|
{
|
|
"before": {
|
|
"status": 200,
|
|
"body": {
|
|
"enabled": False,
|
|
"queue_depth": 0,
|
|
"active_jobs": 0,
|
|
},
|
|
},
|
|
"enter": {"status": [], "body": None},
|
|
"status": {"status": 200, "body": {"enabled": True}},
|
|
}
|
|
)
|
|
|
|
assert error.value.code == "maintenance_response_invalid"
|
|
|
|
|
|
def test_maintenance_queue_precheck_avoids_freezing_queued_work(
|
|
storage_effects, monkeypatch
|
|
):
|
|
token = "test-maintenance-token-value-0123456789"
|
|
calls: list[tuple[str, str]] = []
|
|
|
|
class Response:
|
|
status = 200
|
|
|
|
def __init__(self, payload):
|
|
self.payload = json.dumps(payload).encode("utf-8")
|
|
|
|
def read(self, _limit):
|
|
return self.payload
|
|
|
|
class Connection:
|
|
def __init__(self, _host, *args, **kwargs):
|
|
assert args == (8000,)
|
|
assert kwargs["timeout"] == 5
|
|
|
|
def request(self, method, path, **_kwargs):
|
|
calls.append((method, path))
|
|
|
|
def getresponse(self):
|
|
if len(calls) == 1:
|
|
return Response(
|
|
{
|
|
"enabled": False,
|
|
"queue_depth": 2,
|
|
"active_jobs": 0,
|
|
"active_drained": True,
|
|
"drained": False,
|
|
}
|
|
)
|
|
return Response(
|
|
{
|
|
"enabled": True,
|
|
"queue_depth": 2,
|
|
"active_jobs": 0,
|
|
"active_drained": True,
|
|
"drained": False,
|
|
}
|
|
)
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(http.client, "HTTPConnection", Connection)
|
|
monkeypatch.setattr(sys, "stdin", io.StringIO(token))
|
|
output = io.StringIO()
|
|
with redirect_stdout(output):
|
|
exec( # noqa: S102 - exercise the exact fixed in-container exchange script
|
|
storage_effects._REMOTE_MAINTENANCE_EXCHANGE, {"__name__": "__main__"}
|
|
)
|
|
|
|
response = json.loads(output.getvalue())
|
|
assert calls == [("GET", storage_effects.MAINTENANCE_STATUS_PATH)]
|
|
assert response["enter"] is None
|
|
assert token not in output.getvalue()
|
|
|
|
|
|
def test_legacy_main_revision_mismatch_stops_before_loading_helper(
|
|
storage_effects, tmp_path, monkeypatch
|
|
):
|
|
helper_bytes = b"# reviewed test helper\n"
|
|
bare_repo, actual_revision = _bare_repo_with_helper(tmp_path, helper_bytes)
|
|
source_root = tmp_path / "sources" / "repopulse"
|
|
helper_source = source_root / "scripts" / "repopulse_db_migration.py"
|
|
helper_source.parent.mkdir(parents=True)
|
|
helper_source.write_bytes(helper_bytes)
|
|
expected_revision = "0" * 40
|
|
assert actual_revision != expected_revision
|
|
|
|
core = type("GitCore", (), {})()
|
|
core.repository_path = lambda _repository, _project: bare_repo
|
|
core.output = lambda command: subprocess.run(
|
|
command,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
core.validate_commit = lambda repo, branch, revision: storage_effects._fixed_core(
|
|
None
|
|
).validate_commit(repo, branch, revision)
|
|
|
|
monkeypatch.setattr(storage_effects, "EXACT_SOURCE_ROOT", source_root.parent)
|
|
monkeypatch.setattr(storage_effects._store, "_ROOT_UID", source_root.stat().st_uid)
|
|
|
|
def controller_directory(path: Path, *, root_owned: bool = False):
|
|
storage_effects._store._assert_no_symlink_components(path)
|
|
info = path.lstat()
|
|
assert info.st_mode
|
|
return info
|
|
|
|
monkeypatch.setattr(storage_effects, "_validate_directory", controller_directory)
|
|
monkeypatch.setattr(
|
|
storage_effects,
|
|
"_validate_git_repository_directory",
|
|
lambda path: path.lstat(),
|
|
)
|
|
monkeypatch.setattr(
|
|
storage_effects._store,
|
|
"_safe_directory",
|
|
lambda path, **_kwargs: (
|
|
storage_effects._store._assert_no_symlink_components(path) or path.lstat()
|
|
),
|
|
)
|
|
|
|
with pytest.raises(storage_effects.StorageEffectsError) as error:
|
|
storage_effects._load_reviewed_migration_helper(
|
|
core,
|
|
{"source_dir": str(source_root)},
|
|
expected_revision,
|
|
)
|
|
|
|
assert error.value.code == "main_revision_mismatch"
|
|
|
|
|
|
def test_oversized_helper_is_rejected_before_git_returns_blob(
|
|
storage_effects, tmp_path, monkeypatch
|
|
):
|
|
helper_bytes = b"#" * (storage_effects._HELPER_MAX_BYTES + 1)
|
|
bare_repo, revision = _bare_repo_with_helper(tmp_path, helper_bytes)
|
|
real_run = subprocess.run
|
|
commands: list[list[str]] = []
|
|
|
|
def recording_run(args, *positional, **keyword):
|
|
commands.append(list(args))
|
|
return real_run(args, *positional, **keyword)
|
|
|
|
monkeypatch.setattr(storage_effects.subprocess, "run", recording_run)
|
|
|
|
with pytest.raises(storage_effects.StorageEffectsError) as error:
|
|
storage_effects._read_git_blob(bare_repo, revision)
|
|
|
|
assert error.value.code == "reviewed_migration_helper_invalid"
|
|
assert len(commands) == 1
|
|
assert commands[0][2:4] == ["cat-file", "-s"]
|
|
|
|
|
|
def test_legacy_maintenance_404_blocks_stop(storage_effects, tmp_path, monkeypatch):
|
|
revision = "a" * 40
|
|
container_id = "b" * 64
|
|
image_id = "sha256:" + "c" * 64
|
|
image_ref = "ghcr.io/jens/repopulse:legacy"
|
|
token = "fixed-maintenance-token-value-0123456789"
|
|
intent_id = "rpa-0123456789abcdef0123"
|
|
binding = "d" * 64
|
|
container_id_sha256 = hashlib.sha256(container_id.encode("ascii")).hexdigest()
|
|
image_sha256 = hashlib.sha256(image_ref.encode("utf-8")).hexdigest()
|
|
|
|
class Store:
|
|
def _lock_identity(self):
|
|
return ("project-lock", "global-lock")
|
|
|
|
def _transaction_owner(self, _identity):
|
|
return threading.get_ident()
|
|
|
|
def load_intent(self):
|
|
return {
|
|
"intentId": intent_id,
|
|
"intentBindingSha256": binding,
|
|
"status": "in_progress",
|
|
"phase": "target_gates_ready",
|
|
}
|
|
|
|
class Core:
|
|
def __init__(self):
|
|
self.commands: list[list[str]] = []
|
|
|
|
def inspect_container(self, _name):
|
|
return {
|
|
"Id": container_id,
|
|
"Name": "/repopulse",
|
|
"Image": image_id,
|
|
"Config": {
|
|
"Image": image_ref,
|
|
"Env": [f"REPOPULSE_MAINTENANCE_TOKEN={token}"],
|
|
"Labels": {
|
|
"com.docker.compose.project": "repopulse",
|
|
"org.opencontainers.image.revision": revision,
|
|
},
|
|
},
|
|
"State": {
|
|
"Status": "running",
|
|
"Running": True,
|
|
"Health": {"Status": "healthy"},
|
|
},
|
|
"Mounts": [
|
|
{
|
|
"Type": "volume",
|
|
"Name": "repopulse_data",
|
|
"Destination": "/data",
|
|
"Source": str(storage_effects.LEGACY_VOLUME_MOUNTPOINT),
|
|
"RW": True,
|
|
}
|
|
],
|
|
}
|
|
|
|
def inspect_image(self, _image_id):
|
|
return {
|
|
"Id": image_id,
|
|
"Config": {
|
|
"Labels": {
|
|
"org.opencontainers.image.revision": revision,
|
|
"org.opencontainers.image.source": "jens/repopulse",
|
|
}
|
|
},
|
|
}
|
|
|
|
def run(self, command):
|
|
self.commands.append(list(command))
|
|
|
|
core = Core()
|
|
effects = storage_effects.FixedRepoPulseStorageEffects(
|
|
store=Store(), deployment_core=core
|
|
)
|
|
monkeypatch.setattr(
|
|
effects,
|
|
"_project_and_helper",
|
|
lambda _intent, _revision: ({}, ModuleType("unused"), "e" * 64),
|
|
)
|
|
monkeypatch.setattr(
|
|
storage_effects, "_private_environment_facts", lambda: (token, True, True)
|
|
)
|
|
monkeypatch.setattr(
|
|
storage_effects,
|
|
"_private_record_path",
|
|
lambda _name, _intent: tmp_path / "unused-receipt.json",
|
|
)
|
|
monkeypatch.setattr(storage_effects, "_read_private_record", lambda *_args: None)
|
|
monkeypatch.setattr(
|
|
storage_effects,
|
|
"_assert_legacy_volume_retained",
|
|
lambda _core: storage_effects.LEGACY_VOLUME_MOUNTPOINT,
|
|
)
|
|
monkeypatch.setattr(
|
|
storage_effects,
|
|
"_maintenance_exchange",
|
|
lambda *_args: {
|
|
"before": {"status": 404, "body": None},
|
|
"enter": None,
|
|
"status": None,
|
|
},
|
|
)
|
|
|
|
with pytest.raises(storage_effects.StorageEffectsError) as error:
|
|
effects.ensure_quiesced(
|
|
intent_id=intent_id,
|
|
intent_binding_sha256=binding,
|
|
expected_main_revision=revision,
|
|
expected_container_id_sha256=container_id_sha256,
|
|
expected_container_image_sha256=image_sha256,
|
|
)
|
|
|
|
assert error.value.code == "maintenance_endpoint_unavailable"
|
|
assert core.commands == []
|