115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
"""What a deploy needs from the host running it: this script, ssh, and deploy.env.
|
|
|
|
No checkout of the application repository. The compose files come out of the
|
|
engine image for the tag being deployed, and migrations run from that same image,
|
|
so nothing is rsynced from a working tree any more.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
DEPLOY = Path(__file__).resolve().parent.parent / "deploy.sh"
|
|
|
|
|
|
def _logical_lines() -> list[str]:
|
|
joined = DEPLOY.read_text().replace("\\\n", " ")
|
|
return [line for line in joined.splitlines() if not line.lstrip().startswith("#")]
|
|
|
|
|
|
def _position(needle: str) -> int:
|
|
text = DEPLOY.read_text()
|
|
assert text.count(needle) == 1, f"{needle!r} appears {text.count(needle)} times"
|
|
return text.index(needle)
|
|
|
|
|
|
@pytest.fixture
|
|
def run(tmp_path: Path):
|
|
"""Run the script with no deploy.env and an ssh that records being reached."""
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
marker = tmp_path / "ssh-was-called"
|
|
fake = bin_dir / "ssh"
|
|
fake.write_text(f"#!/bin/sh\ntouch {marker}\nexit 255\n")
|
|
fake.chmod(0o755)
|
|
|
|
def _run(*args: str, **env: str) -> subprocess.CompletedProcess[str]:
|
|
base = {
|
|
"PATH": f"{bin_dir}:{os.environ['PATH']}",
|
|
"HOME": str(tmp_path),
|
|
"DEPLOY_ENV_FILE": "/dev/null",
|
|
}
|
|
result = subprocess.run(
|
|
["bash", str(DEPLOY), *args],
|
|
env={**base, **env},
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
assert not marker.exists(), "the script reached ssh before refusing"
|
|
return result
|
|
|
|
return _run
|
|
|
|
|
|
class TestNoCheckoutNeeded:
|
|
def test_nothing_is_rsynced(self) -> None:
|
|
offenders = [line.strip() for line in _logical_lines() if "rsync" in line]
|
|
assert not offenders, f"deploy still copies files from a working tree: {offenders}"
|
|
|
|
def test_compose_files_come_from_the_engine_image(self) -> None:
|
|
assert "/app/deploy/" in DEPLOY.read_text()
|
|
|
|
def test_compose_files_are_fetched_before_the_pull(self) -> None:
|
|
assert _position("Fetching compose files") < _position("Pulling images...")
|
|
|
|
|
|
class TestTheRegistryPasswordStaysOffCommandLines:
|
|
"""Anything inside the ssh argument becomes the remote shell's command line,
|
|
readable by every user on the host through `ps`. The password travels over
|
|
ssh's stdin instead."""
|
|
|
|
def test_no_ssh_argument_carries_the_password(self) -> None:
|
|
offenders = [
|
|
line.strip()
|
|
for line in _logical_lines()
|
|
if "ssh " in line and "PASS" in line.split("ssh ", 1)[1]
|
|
]
|
|
assert not offenders, f"password on a remote command line: {offenders}"
|
|
|
|
def test_login_reads_the_password_from_stdin(self) -> None:
|
|
assert any("--password-stdin" in line and "| ssh " in line for line in _logical_lines()), (
|
|
"docker login no longer gets the password piped through ssh"
|
|
)
|
|
|
|
|
|
class TestRefusesBeforeTouchingAHost:
|
|
"""Every one of these must stop before the first ssh — the fixture's fake ssh
|
|
fails the test if it is reached."""
|
|
|
|
def test_an_unpublished_version_is_refused(self, run) -> None:
|
|
r = run("--version=bogus", "host", REGISTRY_HOST="registry.example")
|
|
assert r.returncode != 0
|
|
assert "Refusing to deploy version 'bogus'" in r.stderr
|
|
|
|
def test_no_registry_is_refused(self, run) -> None:
|
|
r = run("host")
|
|
assert r.returncode != 0
|
|
assert "REGISTRY_HOST" in r.stderr
|
|
|
|
def test_no_servers_is_refused(self, run) -> None:
|
|
r = run(REGISTRY_HOST="registry.example")
|
|
assert r.returncode != 0
|
|
assert "No servers" in r.stderr
|
|
|
|
@pytest.mark.parametrize("flag", ["--no-cache", "--bogus"])
|
|
def test_an_unknown_flag_is_refused(self, run, flag: str) -> None:
|
|
r = run(flag, "host", REGISTRY_HOST="registry.example")
|
|
assert r.returncode != 0
|
|
assert f"Unknown option: {flag}" in r.stderr
|