"""Unit: bot_bottle.resources — build-resource resolution for both a source checkout and an installed wheel. The checkout path is what the whole test suite already runs under; the wheel path is exercised here by faking an installed layout (a package dir with a bundled ``_resources/`` and no sibling Dockerfiles) and asserting that ``build_root()`` stages a repo-root-shaped context. See ``test_wheel_install.py`` for the end-to-end build+install check. """ from __future__ import annotations import tempfile import unittest from contextlib import contextmanager from pathlib import Path from unittest.mock import patch from bot_bottle import resources from tests.unit import use_bottle_root class TestCheckoutMode(unittest.TestCase): """The environment the suite runs in: a real source checkout.""" def test_is_source_checkout(self): self.assertTrue(resources.is_source_checkout()) def test_build_root_is_repo_root(self): root = resources.build_root() self.assertTrue((root / "bot_bottle").is_dir()) self.assertTrue((root / "pyproject.toml").is_file()) self.assertTrue((root / "Dockerfile.gateway").is_file()) def test_resource_helpers_resolve(self): self.assertTrue(resources.dockerfile("Dockerfile.gateway").is_file()) self.assertTrue(resources.nix_netpool_module().is_file()) self.assertTrue(resources.netpool_script().is_file()) def test_bundled_resources_all_exist_at_root(self): # Drift guard: every path setup.py bundles must exist in the checkout. root = resources.build_root() for rel in resources.BUNDLED_RESOURCES: self.assertTrue((root / rel).is_file(), f"missing bundled resource: {rel}") class TestWheelMode(unittest.TestCase): """Fake an installed wheel: a package dir with _resources/ and no checkout Dockerfiles beside it.""" def _fake_install(self, tmp: Path) -> Path: pkg = tmp / "site-packages" / "bot_bottle" (pkg / "cli").mkdir(parents=True) (pkg / "__init__.py").write_text("") (pkg / "cli" / "__init__.py").write_text("# module\n") bundled = pkg / "_resources" for rel in resources.BUNDLED_RESOURCES: dst = bundled / rel dst.parent.mkdir(parents=True, exist_ok=True) dst.write_text(f"# fake {rel}\n") return pkg @contextmanager def _wheel(self): """Point `resources` at a fake installed wheel with an isolated app-data dir; yields the package dir so a test can mutate it.""" with tempfile.TemporaryDirectory() as tmpname: tmp = Path(tmpname) pkg = self._fake_install(tmp) self.addCleanup(use_bottle_root(tmp / "appdata")) with patch.object(resources, "_PKG", pkg), \ patch.object(resources, "_BUNDLED", pkg / "_resources"), \ patch.object(resources, "_CHECKOUT_ROOT", tmp / "no-checkout"): yield pkg def test_stage_and_resolve(self): with self._wheel(): self.assertFalse(resources.is_source_checkout()) root = resources.build_root() # Staged context looks like a repo root. self.assertTrue((root / "bot_bottle" / "__init__.py").is_file()) self.assertTrue((root / "bot_bottle" / "cli" / "__init__.py").is_file()) self.assertTrue((root / "pyproject.toml").is_file()) self.assertTrue((root / "Dockerfile.gateway").is_file()) self.assertTrue((root / "nix" / "firecracker-netpool.nix").is_file()) self.assertTrue((root / "scripts" / "firecracker-netpool.sh").is_file()) # The bundled-resource copies are NOT re-nested under the staged # package (keeps it byte-identical to a checkout package). self.assertFalse((root / "bot_bottle" / "_resources").exists()) # Helpers resolve off the staged root. self.assertEqual(root / "Dockerfile.gateway", resources.dockerfile("Dockerfile.gateway")) # Idempotent: second call returns the same completed dir. self.assertEqual(root, resources.build_root()) def test_refreshes_when_content_changes_at_same_version(self): # Regression for the stale-cache bug: `pipx install --force` of a newer # commit keeps version 0.1.0, so keying on version would reuse the old # tree. Keying on content must re-stage when a package file changes. with self._wheel() as pkg: root1 = resources.build_root() self.assertTrue((root1 / ".complete").is_file()) (pkg / "cli" / "__init__.py").write_text("# new commit, same version\n") root2 = resources.build_root() self.assertNotEqual(root1, root2) self.assertEqual( "# new commit, same version\n", (root2 / "bot_bottle" / "cli" / "__init__.py").read_text(), ) def test_rebuilds_when_stage_incomplete(self): # A crash mid-stage can leave a dir without its `.complete` marker; the # next call must rebuild it rather than trust the partial tree. with self._wheel(): root = resources.build_root() (root / ".complete").unlink() (root / "sentinel").write_text("stale") again = resources.build_root() self.assertEqual(root, again) # same content digest → same dir self.assertTrue((again / ".complete").is_file()) self.assertFalse((again / "sentinel").exists()) # rebuilt clean def test_reuses_peer_stage_after_lock_wait(self): # Regression for the staging race: a caller that loses the lock must, # once it wins, see the peer's completed tree and reuse it — never # re-clobber a shared path. Drive it deterministically: hold the lock, # let a worker block after its fast-path miss, publish a complete tree # as the "peer", then release so the worker takes the reuse path. import fcntl import threading import time with self._wheel(): base = resources.bot_bottle_root() / "build-root" base.mkdir(parents=True, exist_ok=True) dest = base / resources._content_digest() # pylint: disable=protected-access result: dict[str, Path] = {} with open(base / ".stage.lock", "w", encoding="utf-8") as held: fcntl.flock(held, fcntl.LOCK_EX) def worker() -> None: result["root"] = resources.build_root() t = threading.Thread(target=worker) t.start() # Let the worker miss the fast path (dest not yet complete) and # block on the held lock, then publish a complete tree as a peer # would have and release the lock. time.sleep(0.3) dest.mkdir(parents=True) (dest / ".complete").write_text("") fcntl.flock(held, fcntl.LOCK_UN) t.join(timeout=10) self.assertEqual(dest, result["root"]) self.assertFalse(t.is_alive()) def test_missing_bundle_raises(self): with tempfile.TemporaryDirectory() as tmpname: tmp = Path(tmpname) pkg = tmp / "bot_bottle" pkg.mkdir() restore = use_bottle_root(tmp / "appdata") self.addCleanup(restore) with patch.object(resources, "_PKG", pkg), \ patch.object(resources, "_BUNDLED", pkg / "_resources"), \ patch.object(resources, "_CHECKOUT_ROOT", tmp / "no-checkout"): with self.assertRaises(resources.ResourceError): resources.build_root() if __name__ == "__main__": unittest.main()