166 lines
7.1 KiB
Python
Executable File
166 lines
7.1 KiB
Python
Executable File
#!/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)
|