146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
"""Проверки повторной публикации READY и Git preflight без оборудования."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from tools.rf_handoff_channel import GitResultChannel, ResultChannelError
|
|
|
|
|
|
OLD_HEAD = "1" * 40
|
|
READY_HEAD = "2" * 40
|
|
|
|
|
|
class SimulatedPushChannel(GitResultChannel):
|
|
def __init__(
|
|
self,
|
|
push_results: list[subprocess.CompletedProcess[str]],
|
|
observed_heads: list[str],
|
|
) -> None:
|
|
self.remote = "origin"
|
|
self.branch = "rx-node"
|
|
self.push_results = iter(push_results)
|
|
self.observed_heads = iter(observed_heads)
|
|
self.push_arguments: list[tuple[str, ...]] = []
|
|
self.fetch_calls = 0
|
|
|
|
def _run(
|
|
self,
|
|
arguments: list[str],
|
|
*,
|
|
cwd: Path | None = None,
|
|
check: bool = True,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
del cwd, check
|
|
assert arguments[0] == "push"
|
|
self.push_arguments.append(tuple(arguments))
|
|
return next(self.push_results)
|
|
|
|
def remote_head(self) -> str:
|
|
return next(self.observed_heads)
|
|
|
|
def fetch(self) -> str:
|
|
self.fetch_calls += 1
|
|
return OLD_HEAD
|
|
|
|
|
|
def _push_result(returncode: int, stderr: str = "") -> subprocess.CompletedProcess[str]:
|
|
return subprocess.CompletedProcess(["git", "push"], returncode, "", stderr)
|
|
|
|
|
|
def _git(arguments: list[str], cwd: Path) -> str:
|
|
return subprocess.run(
|
|
["git", *arguments],
|
|
cwd=cwd,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
check=True,
|
|
).stdout.strip()
|
|
|
|
|
|
def test_transient_failure_retries_the_same_ready_commit_once() -> None:
|
|
channel = SimulatedPushChannel(
|
|
[_push_result(1, "Authentication failed"), _push_result(0)],
|
|
[OLD_HEAD, READY_HEAD],
|
|
)
|
|
assert channel._push_exact_commit(Path("checkout"), READY_HEAD, OLD_HEAD) == READY_HEAD
|
|
assert len(channel.push_arguments) == 2
|
|
assert channel.push_arguments[0] == channel.push_arguments[1]
|
|
assert channel.push_arguments[0][-1] == f"{READY_HEAD}:refs/heads/rx-node"
|
|
assert channel.fetch_calls == 1
|
|
|
|
|
|
def test_ambiguous_push_is_success_when_remote_already_has_ready() -> None:
|
|
channel = SimulatedPushChannel(
|
|
[_push_result(1, "TLS connection was non-properly terminated")],
|
|
[READY_HEAD],
|
|
)
|
|
assert channel._push_exact_commit(Path("checkout"), READY_HEAD, OLD_HEAD) == READY_HEAD
|
|
assert len(channel.push_arguments) == 1
|
|
assert channel.fetch_calls == 0
|
|
|
|
|
|
def test_second_transient_failure_does_not_create_another_ready() -> None:
|
|
channel = SimulatedPushChannel(
|
|
[
|
|
_push_result(1, "Authentication failed"),
|
|
_push_result(1, "Authentication failed"),
|
|
],
|
|
[OLD_HEAD, OLD_HEAD],
|
|
)
|
|
with pytest.raises(ResultChannelError, match="after one retry"):
|
|
channel._push_exact_commit(Path("checkout"), READY_HEAD, OLD_HEAD)
|
|
assert len(channel.push_arguments) == 2
|
|
assert channel.push_arguments[0] == channel.push_arguments[1]
|
|
|
|
|
|
def test_failed_write_preflight_stops_before_any_publication() -> None:
|
|
class FailingPreflight(GitResultChannel):
|
|
def __init__(self) -> None:
|
|
self.remote = "origin"
|
|
self.branch = "rx-node"
|
|
|
|
def fetch(self) -> str:
|
|
return OLD_HEAD
|
|
|
|
def remote_head(self) -> str:
|
|
raise AssertionError("remote confirmation must not follow failed dry-run")
|
|
|
|
def _run(
|
|
self,
|
|
arguments: list[str],
|
|
*,
|
|
cwd: Path | None = None,
|
|
check: bool = True,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
del cwd, check
|
|
assert arguments[:3] == ["push", "--dry-run", "--porcelain"]
|
|
return _push_result(1, "Authentication failed")
|
|
|
|
with pytest.raises(ResultChannelError, match="push preflight failed"):
|
|
FailingPreflight().verify_write_access()
|
|
|
|
|
|
def test_write_preflight_checks_access_without_changing_remote(tmp_path: Path) -> None:
|
|
remote = tmp_path / "remote.git"
|
|
work = tmp_path / "work"
|
|
_git(["init", "--bare", str(remote)], tmp_path)
|
|
work.mkdir()
|
|
_git(["init"], work)
|
|
_git(["config", "user.name", "Test"], work)
|
|
_git(["config", "user.email", "test@example.invalid"], work)
|
|
(work / "seed.txt").write_text("seed\n", encoding="utf-8")
|
|
_git(["add", "seed.txt"], work)
|
|
_git(["commit", "-m", "Seed result branch"], work)
|
|
_git(["remote", "add", "origin", str(remote)], work)
|
|
_git(["push", "origin", "HEAD:refs/heads/rx-node"], work)
|
|
channel = GitResultChannel(work)
|
|
before = channel.remote_head()
|
|
assert channel.verify_write_access() == before
|
|
assert channel.remote_head() == before
|