89 lines
4.9 KiB
Python
89 lines
4.9 KiB
Python
"""Exercise the workflow's real rsync script with a mocked service manager."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
|
|
|
|
@unittest.skipUnless(shutil.which("rsync") and shutil.which("flock"), "deployment tools are not installed")
|
|
class DeploymentTests(unittest.TestCase):
|
|
def test_deploy_preserves_runtime_and_archive_and_restarts_service(self):
|
|
self._run_deployment(archive_exists=True)
|
|
|
|
def test_deploy_creates_archive_before_restarting_service(self):
|
|
self._run_deployment(archive_exists=False)
|
|
|
|
def _run_deployment(self, *, archive_exists):
|
|
workflow = (Path(__file__).resolve().parents[1] / ".gitea/workflows/deploy.yml").read_text()
|
|
deploy_step = workflow.split("- name: Deploy and restart\n", 1)[1]
|
|
script = textwrap.dedent(deploy_step.split("run: |\n", 1)[1])
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
checkout, deployment, binaries = [root / name for name in ("checkout", "deployment", "bin")]
|
|
for path in (checkout, deployment, binaries):
|
|
path.mkdir()
|
|
(checkout / "solar_relay").mkdir()
|
|
(checkout / "solar_relay" / "new.py").write_text("# new release\n")
|
|
(checkout / "requirements.txt").write_text("pyarrow==26.0.0\n")
|
|
(deployment / "obsolete.py").write_text("# obsolete\n")
|
|
protected = {
|
|
"config.toml": "private configuration",
|
|
".env": "private token",
|
|
"data/archive/date=2026-11-01/samples.parquet": "durable parquet observations",
|
|
"archive/date=2026-11-01/samples.parquet": "archived observations",
|
|
".venv/keep.txt": "runtime venv",
|
|
".git/keep.txt": "existing checkout metadata",
|
|
}
|
|
if not archive_exists:
|
|
del protected["data/archive/date=2026-11-01/samples.parquet"]
|
|
for relative, content in protected.items():
|
|
target = deployment / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(content)
|
|
source = checkout / relative
|
|
source.parent.mkdir(parents=True, exist_ok=True)
|
|
source.write_text("must not replace local state")
|
|
runtime = deployment / ".venv/bin/python"
|
|
runtime.parent.mkdir(parents=True)
|
|
runtime.write_text('#!/bin/sh\nif [ "$1" = "-m" ] && [ "$2" = pip ]; then printf "pip %s\\n" "$*" >> "$SOLAR_DEPLOY_TEST_LOG"; exit 0; fi\nexec ' + shlex.quote(sys.executable) + ' "$@"\n')
|
|
runtime.chmod(0o755)
|
|
mocks = {
|
|
"sudo": '#!/bin/sh\n[ "$1" = "-n" ] || exit 99\nshift\nexec "$@"\n',
|
|
"systemctl": '#!/bin/sh\nif [ "$1" = restart ] && [ ! -d "$DEPLOY_PATH/data/archive" ]; then echo "Archive directory is missing before service restart" >&2; exit 1; fi\nprintf "%s\\n" "$*" >> "$SOLAR_DEPLOY_TEST_LOG"\nif [ "$1" = show ]; then printf "0\\n"; fi\n',
|
|
"sleep": "#!/bin/sh\nexit 0\n",
|
|
"curl": '#!/bin/sh\nprintf "curl %s\\n" "$*" >> "$SOLAR_DEPLOY_TEST_LOG"\n',
|
|
}
|
|
for name, content in mocks.items():
|
|
target = binaries / name
|
|
target.write_text(content)
|
|
target.chmod(0o755)
|
|
log = root / "systemctl.log"
|
|
env = {**os.environ, "DEPLOY_PATH": str(deployment), "SERVICE": "solar-relay.service",
|
|
"PATH": str(binaries) + os.pathsep + os.environ["PATH"], "SOLAR_DEPLOY_TEST_LOG": str(log),
|
|
"HEALTH_URL": "https://rpioffice.nest.lutran.ch/solar-relay/healthz"}
|
|
result = subprocess.run(["bash", "-c", script], cwd=checkout, env=env, capture_output=True, text=True, timeout=15)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue((deployment / "data/archive").is_dir())
|
|
if not archive_exists:
|
|
for relative in ("data", "data/archive"):
|
|
self.assertEqual((deployment / relative).stat().st_mode & 0o777, 0o700)
|
|
self.assertTrue((deployment / "solar_relay/new.py").exists())
|
|
self.assertFalse((deployment / "obsolete.py").exists())
|
|
for relative, content in protected.items():
|
|
self.assertEqual((deployment / relative).read_text(), content, relative)
|
|
self.assertIn("restart solar-relay.service", log.read_text())
|
|
self.assertIn("is-active --quiet solar-relay.service", log.read_text())
|
|
self.assertIn("pip -m pip install --only-binary=pyarrow -r", log.read_text())
|
|
self.assertLess(log.read_text().index("pip -m pip"), log.read_text().index("restart solar-relay.service"))
|
|
self.assertIn(env["HEALTH_URL"], log.read_text())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|