Initial import of the netOrk deploy script

tests/test_deploy_script_bootstrap.py

@@ -0,0 +1,114 @@
+"""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