Digital Drosophila Archive
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""Synthetic temporary fixtures and mocked Git/content operations; no transfer."""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import download_verify as D
|
|
|
|
|
|
class DownloadTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.checkout = self.base / "verification"
|
|
(self.checkout / ".git").mkdir(parents=True)
|
|
(self.checkout / "registry").mkdir()
|
|
(self.checkout / D.PAYLOAD).mkdir(parents=True)
|
|
self.cache = self.checkout / ".git" / "brand-new-lfs"
|
|
self.receipt = self.checkout / "registry" / "transport.json"
|
|
self.content_receipt = self.checkout / "registry" / "content.json"
|
|
self.history = self.base / "history.gz"
|
|
self.history.write_bytes(b"synthetic history placeholder")
|
|
self.commit = "a" * 40
|
|
self.calls = []
|
|
self.pull_code = self.fsck_code = 0
|
|
self.remote = D.REMOTE
|
|
self.head = self.commit
|
|
self.dirty = ""
|
|
self.content_failure = False
|
|
self.config_exit = 1
|
|
self.adapter_config = {}
|
|
self.local_adapter_config = {}
|
|
self.adapter_blob = b"# Synthetic LAN adapter fixture; never executed.\n"
|
|
self.addCleanup(patch.stopall)
|
|
patch.object(D.Path, "cwd", return_value=self.checkout).start()
|
|
patch.object(D, "command", side_effect=self.command).start()
|
|
self.verifier = patch.object(D.content, "verify_publication", side_effect=self.verify).start()
|
|
self.capture = contextlib.redirect_stdout(io.StringIO())
|
|
self.capture.__enter__()
|
|
self.addCleanup(self.capture.__exit__, None, None, None)
|
|
|
|
def command(self, args, *, text=False, bytes_output=False):
|
|
self.calls.append(args)
|
|
if "rev-parse" in args:
|
|
if "HEAD" in args:
|
|
return self.head
|
|
return str(self.checkout / ".git") if "--absolute-git-dir" in args else str(self.checkout)
|
|
if "remote" in args:
|
|
return self.remote
|
|
if "status" in args:
|
|
return self.dirty
|
|
if "--get-regexp" in args:
|
|
if "--name-only" in args:
|
|
return "\n".join(self.adapter_config)
|
|
return self.config_exit
|
|
if "--get-all" in args:
|
|
values = self.local_adapter_config if "--local" in args else self.adapter_config
|
|
return values.get(args[-1], "")
|
|
if "show" in args:
|
|
return self.adapter_blob
|
|
if "pull" in args:
|
|
oid = "f" * 64
|
|
obj = self.cache / "objects" / oid[:2] / oid[2:4] / oid
|
|
obj.parent.mkdir(parents=True)
|
|
obj.write_bytes(b"synthetic object")
|
|
return self.pull_code
|
|
if "fsck" in args:
|
|
return self.fsck_code
|
|
return "synthetic-version"
|
|
|
|
def verify(self, root, receipt, history):
|
|
self.assertEqual(root, self.checkout / D.PAYLOAD)
|
|
self.assertEqual(history, self.history)
|
|
if self.content_failure:
|
|
raise ValueError("synthetic failure")
|
|
D.archives.write_new(receipt, b'{"synthetic":true}\n')
|
|
return {"pass_all": True, "receipt_sha256": D.archives.sha_file(receipt)["sha256"],
|
|
"aggregate_inventory_sha256": "b" * 64, "historical_status": "FULL_CURRENT_MEMBER_MATCH"}
|
|
|
|
def run_attempt(self):
|
|
return D.run(self.commit, self.cache, self.receipt, self.content_receipt, self.history)
|
|
|
|
def prepare_adapter(self):
|
|
adapter = self.checkout / "tools/lfs-publication/lan_lfs_transfer.py"
|
|
adapter.parent.mkdir(parents=True)
|
|
adapter.write_bytes(self.adapter_blob)
|
|
prefix = "lfs.customtransfer.malecns-lan-https."
|
|
self.adapter_config = {"lfs.standalonetransferagent": "malecns-lan-https",
|
|
prefix + "path": "/usr/bin/python3", prefix + "args": "-B " + str(adapter),
|
|
prefix + "concurrent": "true", prefix + "direction": "both"}
|
|
self.local_adapter_config = self.adapter_config.copy()
|
|
self.config_exit = 0
|
|
return adapter
|
|
|
|
def assert_not_started(self):
|
|
self.assertFalse(self.cache.exists())
|
|
self.assertFalse(self.receipt.exists())
|
|
self.assertFalse(any("pull" in args for args in self.calls))
|
|
|
|
def test_success_binds_new_cache_exits_content_and_source_sha(self):
|
|
result = self.run_attempt()
|
|
self.assertEqual(json.loads(self.receipt.read_bytes()), result)
|
|
self.assertTrue(result["cache"]["initially_absent"])
|
|
self.assertEqual(result["cache"]["before"], {"objects": 0, "bytes": 0})
|
|
self.assertEqual(result["cache"]["after"], {"objects": 1, "bytes": 16})
|
|
self.assertEqual(result["content_receipt"]["sha256"], D.archives.sha_file(self.content_receipt)["sha256"])
|
|
self.assertEqual(len(result["tool_sources"]), 3)
|
|
self.assertFalse(result["learning_pass_claimed"])
|
|
pull = next(args for args in self.calls if "pull" in args)
|
|
fsck = next(args for args in self.calls if "fsck" in args)
|
|
self.assertIn("--include=artifacts/**", pull)
|
|
self.assertIn("--exclude=", pull)
|
|
self.assertEqual(pull[-1], "origin")
|
|
self.assertIn("--dry-run", fsck)
|
|
for args in (pull, fsck):
|
|
self.assertIn("lfs.storage=" + str(self.cache), args)
|
|
self.assertIn("lfs.concurrenttransfers=2", args)
|
|
|
|
def test_existing_cache_receipt_and_content_receipt_are_never_reused(self):
|
|
for path in (self.cache, self.receipt, self.content_receipt):
|
|
with self.subTest(path=path):
|
|
path.write_bytes(b"keep")
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assertEqual(path.read_bytes(), b"keep")
|
|
self.assertFalse(any("pull" in args for args in self.calls))
|
|
path.unlink()
|
|
|
|
def test_wrong_head_or_remote_or_dirty_tree_refused(self):
|
|
for attr, value in (("head", "c" * 40), ("remote", "https://wrong.invalid/repo.git"), ("dirty", " M file")):
|
|
original = getattr(self, attr)
|
|
setattr(self, attr, value)
|
|
with self.subTest(attr=attr), self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assert_not_started()
|
|
setattr(self, attr, original)
|
|
|
|
def test_unsafe_paths_refused_before_transfer(self):
|
|
original = self.cache
|
|
for unsafe in (self.checkout / "payload-cache", self.base / ".." / "escape",
|
|
D.PROTECTED / "should-never-create"):
|
|
self.cache = unsafe
|
|
with self.subTest(path=unsafe), self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assertFalse(any("pull" in args for args in self.calls))
|
|
self.cache = original
|
|
self.receipt = self.checkout / D.PAYLOAD / "receipt.json"
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assert_not_started()
|
|
|
|
def test_pull_failure_retains_partial_and_never_fscks_or_retries(self):
|
|
self.pull_code = 9
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assertTrue(self.cache.exists())
|
|
self.assertFalse(self.receipt.exists())
|
|
self.assertFalse(self.content_receipt.exists())
|
|
self.assertFalse(any("fsck" in args for args in self.calls))
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assertEqual(sum("pull" in args for args in self.calls), 1)
|
|
|
|
def test_endpoint_and_custom_transfer_overrides_fail_closed(self):
|
|
for code in (0, 2):
|
|
self.config_exit = code
|
|
with self.subTest(code=code), self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assert_not_started()
|
|
self.config_exit = 1
|
|
(self.checkout / ".lfsconfig").write_text("[lfs]\nurl = https://not-approved.invalid\n")
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assert_not_started()
|
|
|
|
def test_optional_lan_adapter_is_exact_local_and_head_bound(self):
|
|
adapter = self.prepare_adapter()
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assert_not_started()
|
|
report = D.run(self.commit, self.cache, self.receipt, self.content_receipt,
|
|
self.history, allow_lan_adapter=True)
|
|
self.assertTrue(report["lan_adapter"]["enabled"])
|
|
self.assertEqual(report["tool_sources"]["lan_lfs_transfer.py"]["sha256"],
|
|
D.archives.sha_file(adapter)["sha256"])
|
|
|
|
def test_lan_wrong_blob_global_override_or_other_endpoint_rejected(self):
|
|
self.prepare_adapter()
|
|
self.adapter_blob = b"not the actual file"
|
|
with self.assertRaises(ValueError):
|
|
D.transfer_policy(self.checkout, self.commit, True)
|
|
self.adapter_blob = (self.checkout / "tools/lfs-publication/lan_lfs_transfer.py").read_bytes()
|
|
original = self.local_adapter_config["lfs.standalonetransferagent"]
|
|
self.local_adapter_config["lfs.standalonetransferagent"] = ""
|
|
with self.assertRaises(ValueError):
|
|
D.transfer_policy(self.checkout, self.commit, True)
|
|
self.local_adapter_config["lfs.standalonetransferagent"] = original
|
|
self.adapter_config["lfs.url"] = "https://wrong.invalid"
|
|
with self.assertRaises(ValueError):
|
|
D.transfer_policy(self.checkout, self.commit, True)
|
|
self.assert_not_started()
|
|
|
|
def test_fsck_failure_does_not_call_content_or_publish_receipt(self):
|
|
self.fsck_code = 8
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.verifier.assert_not_called()
|
|
self.assertTrue(self.cache.exists())
|
|
self.assertFalse(self.receipt.exists())
|
|
|
|
def test_content_failure_leaves_no_transport_pass(self):
|
|
self.content_failure = True
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assertFalse(self.receipt.exists())
|
|
self.assertEqual(sum("pull" in args for args in self.calls), 1)
|
|
|
|
def test_head_changed_during_verification_preserves_content_but_no_transport_pass(self):
|
|
def changed(*args):
|
|
result = self.verify(*args)
|
|
self.head = "d" * 40
|
|
return result
|
|
self.verifier.side_effect = changed
|
|
with self.assertRaises(ValueError):
|
|
self.run_attempt()
|
|
self.assertTrue(self.content_receipt.exists())
|
|
self.assertFalse(self.receipt.exists())
|
|
|
|
def test_cli_failure_never_prints_exception_diagnostics(self):
|
|
with patch.object(D, "run", side_effect=RuntimeError("secret should not escape")), \
|
|
patch.object(D.sys, "argv", ["download_verify", "--expected-commit", self.commit,
|
|
"--cache", str(self.cache), "--receipt", str(self.receipt),
|
|
"--content-receipt", str(self.content_receipt), "--historical-inventory", str(self.history)]), \
|
|
contextlib.redirect_stdout(io.StringIO()) as output:
|
|
self.assertEqual(D.main(), 1)
|
|
self.assertNotIn("secret should not escape", output.getvalue())
|
|
self.assertIn("FAILED-NOT-RETRIED", output.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|