"""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()