Initial commit

tests/test_deploy.py

@@ -0,0 +1,76 @@
+"""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):
+        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",
+            }
+            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\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 / "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()