#!/usr/bin/env python3 """Tests for felhom-release-copy (D33). No network, no registry, no ep0: a synthetic skopeo `dir:` copy, and the end-to-end refusals through main() with the copy steps replaced. Run: python3 test_release_copy.py""" import hashlib import importlib.machinery import importlib.util import json import os import tempfile import unittest from unittest import mock HERE = os.path.dirname(os.path.abspath(__file__)) _loader = importlib.machinery.SourceFileLoader("rc", os.path.join(HERE, "felhom-release-copy")) _spec = importlib.util.spec_from_loader("rc", _loader) rc = importlib.util.module_from_spec(_spec) _loader.exec_module(rc) def h(b): return hashlib.sha256(b).hexdigest() def make_image(d): """A two-level image like gitea's: an OCI index → one platform manifest → config + one layer. Returns the digest.""" os.makedirs(d) layer, config = b"layer-bytes", b'{"config":1}' child = json.dumps({"schemaVersion": 2, "config": {"digest": "sha256:" + h(config)}, "layers": [{"digest": "sha256:" + h(layer)}]}).encode() index = json.dumps({"schemaVersion": 2, "manifests": [{"digest": "sha256:" + h(child)}]}).encode() for name, data in ((h(layer), layer), (h(config), config), (h(child) + ".manifest.json", child), ("manifest.json", index), ("version", b"Directory Transport Version: 1.1\n")): open(os.path.join(d, name), "wb").write(data) return "sha256:" + h(index) class Verify(unittest.TestCase): def setUp(self): self.t = tempfile.mkdtemp() self.d = os.path.join(self.t, "img") self.digest = make_image(self.d) def test_a_true_copy_passes(self): rc.verify_image_dir(self.d, self.digest) def test_a_changed_layer_is_refused(self): # RED-PROOF: drop the per-blob hash check in verify_image_dir → this passes silently blob = [n for n in os.listdir(self.d) if rc.HEX64.match(n)][0] with open(os.path.join(self.d, blob), "ab") as f: f.write(b"x") with self.assertRaisesRegex(rc.Refused, "does not hash to its name"): rc.verify_image_dir(self.d, self.digest) def test_another_image_is_refused(self): with self.assertRaisesRegex(rc.Refused, "!= registry"): rc.verify_image_dir(self.d, "sha256:" + "0" * 64) def test_a_missing_layer_is_refused(self): # RED-PROOF: drop the „named by a manifest is missing" check → a copy without its layer passes layer = [n for n in os.listdir(self.d) if rc.HEX64.match(n) and open(os.path.join(self.d, n), "rb").read() == b"layer-bytes"][0] os.remove(os.path.join(self.d, layer)) with self.assertRaisesRegex(rc.Refused, "missing"): rc.verify_image_dir(self.d, self.digest) def test_a_missing_child_manifest_is_refused(self): child = [n for n in os.listdir(self.d) if n.endswith(".manifest.json")][0] os.remove(os.path.join(self.d, child)) with self.assertRaisesRegex(rc.Refused, "child manifest .* missing"): rc.verify_image_dir(self.d, self.digest) class Select(unittest.TestCase): def test_newest_releases_by_number_not_text(self): tags = ["0.9.0", "0.10.0", "0.340.0", "0.339.0", "0.340.0-rc.8", "latest", "0.99.1"] self.assertEqual(rc.newest(tags, 3), ["0.99.1", "0.339.0", "0.340.0"]) def test_spoiled_tags_are_never_copied(self): self.assertEqual(rc.newest(["0.314.0", "0.315.0", "0.316.3"], 3, rc.SPOILED["felhom-controller"]), ["0.314.0"]) def test_prune_deletes_beyond_keep(self): t = tempfile.mkdtemp() for v in ("0.1.0", "0.2.0", "0.3.0"): os.makedirs(os.path.join(t, v)) self.assertEqual(rc.prune(t, {"0.3.0"}), ["0.1.0", "0.2.0"]) self.assertEqual(os.listdir(t), ["0.3.0"]) class Main(unittest.TestCase): """main() with the registry copies replaced: the cap and the post-push comparison gate the success signal.""" def setUp(self): self.t = tempfile.mkdtemp() self.conf, self.state, self.tf = (os.path.join(self.t, n) for n in ("conf", "state", "textfile")) for d in (self.conf, self.state, self.tf): os.makedirs(d) open(os.path.join(self.conf, "env"), "w").write( "REMOTE=felhom-relcopy@ep0\nKEEP=3\nAGENT_KEEP=5\nCAP_MB=1\nREPOS='felhom-hub'\n") open(os.path.join(self.conf, "registry-token"), "w").write("not-a-token\n") self.patches = [mock.patch.object(rc, "CONF", self.conf), mock.patch.object(rc, "STATE", self.state), mock.patch.object(rc, "TEXTFILE_DIR", self.tf)] for p in self.patches: p.start() self.addCleanup(lambda: [p.stop() for p in self.patches]) self.pushed = [] def fake_images(self, size): def f(authfile, stage, repos, keep): d = os.path.join(stage, "images", "felhom-hub") os.makedirs(d, exist_ok=True) open(os.path.join(d, "blob"), "wb").write(b"x" * size) return ["felhom-hub:0.167.0"] return f def fake_rsync(self, diff=""): def f(env, stage, dry_check=False): self.pushed.append(dry_check) return diff if dry_check else "" return f def go(self, size, diff=""): with mock.patch.object(rc, "copy_images", self.fake_images(size)), \ mock.patch.object(rc, "copy_agents", lambda reg, stage, keep: None), \ mock.patch.object(rc, "rsync_to", self.fake_rsync(diff)): rc.main() def signal(self): return os.path.exists(os.path.join(self.tf, "felhom_release_copy.prom")) def test_success_pushes_compares_then_signals(self): self.go(1000) self.assertEqual(self.pushed, [False, True]) self.assertTrue(self.signal()) sums = open(os.path.join(self.state, "stage", "SHA256SUMS")).read() self.assertIn("images/felhom-hub/blob", sums) def test_above_the_cap_nothing_is_sent(self): # RED-PROOF: drop the cap check in main() → the push happens and the signal is written with self.assertRaisesRegex(rc.Refused, "above the cap"): self.go(2 * 1024 * 1024) self.assertEqual(self.pushed, []) self.assertFalse(self.signal()) def test_a_difference_after_the_push_writes_no_signal(self): # RED-PROOF: ignore the dry-run's output → a copy ep0 does not hold reads as success with self.assertRaisesRegex(rc.Refused, "differs"): self.go(1000, diff=">fcsT...... images/felhom-hub/blob\n") self.assertFalse(self.signal()) class Units(unittest.TestCase): def read(self, n): return open(os.path.join(HERE, n)).read() def test_unit_runs_the_installed_script_and_mails_on_failure(self): s = self.read("felhom-release-copy.service") self.assertIn("ExecStart=/usr/local/sbin/felhom-release-copy\n", s) self.assertIn("OnFailure=felhom-backup-failmail@%n.service", s) def test_nightly(self): self.assertIn("OnCalendar=*-*-* 01:40:00", self.read("felhom-release-copy.timer")) if __name__ == "__main__": unittest.main(verbosity=2)