Something went wrong. Try again.
Personal Extensions and Nix integration for Albedo.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480"""Deterministic subprocess races and caps not reliably driven through model E2E.
Set ALBEDO_FORMATTER_SOURCE to the integrated upstream priv/python directory."""
import asyncioimport importlibimport jsonimport osimport signalimport shutilimport sysimport tempfileimport typesimport unittestfrom pathlib import Pathfrom unittest.mock import patch
sys.path.insert(0, os.environ["ALBEDO_FORMATTER_SOURCE"])formatter = importlib.import_module("albedo_formatter")plugin = importlib.import_module("albedo_plugins.formatter")files_plugin = importlib.import_module("albedo_plugins.files")
class FormatterTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name).resolve() self.script = Path(__file__).parent / "fixtures" / "formatter-unit.py" self.control = self.root / "formatter-state.json" self.set_fixture("uppercase") self.config = { "formatters": { "upper": { "command": [sys.executable, str(self.script), str(self.control)] } }, "chains": {".py": ["upper"]}, "timeoutMs": 2000, "treefmt": [sys.executable, str(self.script), str(self.control)], "nix": [sys.executable, str(self.script), str(self.control)], } formatter.configure(self.config, str(self.root)) self.fs = files_plugin.Files()
def set_fixture(self, mode, *, control=None, **parameters): state_file = control or self.control state_file.write_text(json.dumps({"mode": mode, **parameters}))
def write(self, name="example.py", text="lower\n"): path = self.root / name self.fs.write(str(path), text) return path
async def test_no_immediate_formatting_and_mode(self): path = self.write() path.chmod(0o751) formatter.record(path) self.assertEqual(path.read_text(), "lower\n") result = await formatter.finish() self.assertTrue(result["ok"], result["error"]) self.assertEqual(result, {"ok": True, "changed": True, "error": ""}) self.assertEqual(path.read_text(), "LOWER\n") self.assertEqual(path.stat().st_mode & 0o777, 0o751) self.assertFalse(formatter._pending)
async def test_edit_hook_and_failed_edit(self): path = self.write() await formatter.finish() self.fs.edit(str(path), "LOWER", "other") self.assertEqual(path.read_text(), "other\n") before = formatter._pending.copy() with self.assertRaises(ValueError): self.fs.edit(str(path), "absent", "new") self.assertEqual(formatter._pending, before) self.assertTrue((await formatter.finish())["changed"]) self.assertEqual(path.read_text(), "OTHER\n")
async def test_unchanged_and_unsupported(self): self.write(text="UPPER\n") other = self.write("notes.txt") self.assertEqual( await formatter.finish(), {"ok": True, "changed": False, "error": ""} ) self.assertEqual(other.read_text(), "lower\n") self.assertFalse(formatter._pending)
async def test_failed_command_keeps_original_and_retry(self): path = self.write() self.set_fixture("failure", diagnostic="broken", exit_code=2) result = await formatter.finish() self.assertFalse(result["ok"]) self.assertIn("broken", result["error"]) self.assertEqual(path.read_text(), "lower\n") self.assertIn(path, formatter._pending) self.set_fixture("write", text="fixed") self.assertTrue((await formatter.finish())["ok"]) self.assertEqual(path.read_text(), "fixed")
async def test_concurrent_user_change(self): path = self.write() self.set_fixture("concurrent_change", original=str(path)) result = await formatter.finish() self.assertFalse(result["ok"]) self.assertIn("concurrent", result["error"]) self.assertEqual(path.read_text(), "user change") self.assertIn(path, formatter._pending) self.assertFalse(list(self.root.glob("albedo-format-*"))) self.fs.write(str(path), "repair") self.set_fixture("write", text="fixed") self.assertTrue((await formatter.finish())["ok"])
async def test_outside_ignored_but_symlink_escape_fails(self): with tempfile.TemporaryDirectory() as outside: path = Path(outside) / "outside.py" self.fs.write(str(path), "outside") result = await formatter.finish() self.assertTrue(result["ok"]) self.assertFalse(formatter._pending) self.assertEqual(path.read_text(), "outside") formatter.configure(self.config, str(self.root)) link = self.root / "link.py" link.symlink_to(path) self.fs.write(str(link), "new outside") self.assertFalse((await formatter.finish())["ok"])
async def test_replaced_original_identity_rejected(self): path = self.write() path.unlink() path.write_text("lower\n") self.assertFalse((await formatter.finish())["ok"]) self.assertEqual(path.read_text(), "lower\n")
async def test_limits_and_output(self): with patch.object(formatter, "MAX_FILES", 1): self.write() self.write("other.py") self.assertEqual(len(formatter._pending), 1) self.assertFalse((await formatter.finish())["ok"]) self.assertFalse((await formatter.finish())["ok"]) formatter.configure(self.config, str(self.root)) with patch.object(formatter, "MAX_BYTES", 3): self.write() self.assertFalse((await formatter.finish())["ok"]) formatter.configure(self.config, str(self.root)) self.write() self.set_fixture("verbose_failure") result = await formatter.finish() self.assertFalse(result["ok"]) self.assertLessEqual(len(result["error"]), 16384)
async def test_timeout_cleanup(self): path = self.write() self.config["timeoutMs"] = 50 self.set_fixture("timeout") result = await formatter.finish() self.assertFalse(result["ok"]) self.assertIn("timed out", result["error"]) self.assertEqual(path.read_text(), "lower\n") self.assertFalse(list(self.root.glob("albedo-format-*")))
async def test_cancellation_kills_group_and_preserves_original(self): path = self.write() started = self.root / "started" self.set_fixture("blocked", started=str(started)) task = asyncio.create_task(formatter.finish()) async with asyncio.timeout(2): while not started.exists(): await asyncio.sleep(0.005) pid = int(started.read_text()) task.cancel() with self.assertRaises(asyncio.CancelledError): await task with self.assertRaises(ProcessLookupError): os.kill(pid, signal.SIGCONT) self.assertEqual(path.read_text(), "lower\n") self.assertIn(path, formatter._pending) self.assertFalse(list(self.root.glob("albedo-format-*")))
async def test_project_scope_treefmt_and_flake(self): for project in ("treefmt.toml", ".treefmt.toml", "flake.nix"): with self.subTest(project=project): formatter.configure(self.config, str(self.root)) marker = self.root / project marker.write_text("") self.addCleanup(marker.unlink, missing_ok=True) path = self.write() log = self.root / "argv.json" self.set_fixture("project", log=str(log)) self.assertTrue((await formatter.finish())["ok"]) argv = json.loads(log.read_text()) self.assertIn("--", argv) staged = Path(argv[-1]) self.assertEqual(staged.name, path.name) self.assertEqual(staged.parent.name, "workspace") self.assertNotEqual(staged, path) if project == "flake.nix": self.assertEqual(argv[:2], ["fmt", "--"]) else: self.assertIn("--no-cache", argv) self.assertIn(str(staged.parent / project), argv) marker.unlink()
@unittest.skipUnless(shutil.which("treefmt"), "standalone test requires treefmt") async def test_real_treefmt_exact_paths_and_config_context(self): treefmt = shutil.which("treefmt") assert treefmt is not None self.config["treefmt"] = [treefmt] (self.root / "src").mkdir() (self.root / "src" / "style.txt").write_text("context-formatted\n") (self.root / "config").mkdir() (self.root / "config" / "style.txt").write_text("context-formatted\n") script = self.root / "config" / "format.py" script.write_text( "import pathlib, sys\n" "for name in sys.argv[1:]:\n" " p = pathlib.Path(name)\n" " assert (p.parent / 'style.txt').read_text() == 'context-formatted\\n'\n" " p.write_text(pathlib.Path('config/style.txt').read_text())\n" ) (self.root / "treefmt.toml").write_text( "[formatter.exact]\n" f"command = {json.dumps(sys.executable)}\n" 'options = ["config/format.py"]\n' 'includes = ["src/example.py", "src/excluded.py"]\n' 'excludes = ["src/excluded.py"]\n' ) # Unrelated development/dependency trees must not be traversed. (self.root / ".direnv").symlink_to("/nix/store", target_is_directory=True) (self.root / "node_modules").mkdir() (self.root / "node_modules" / "cycle").symlink_to(".", target_is_directory=True) path = self.write("src/example.py") excluded = self.write("src/excluded.py") unrelated = self.root / "src" / "dirty.py" unrelated.write_text("unrelated dirty\n") result = await formatter.finish() self.assertEqual(result, {"ok": True, "changed": True, "error": ""}) self.assertEqual(path.read_text(), "context-formatted\n") self.assertEqual(excluded.read_text(), "lower\n") self.assertEqual(unrelated.read_text(), "unrelated dirty\n") self.assertFalse(formatter._pending) # A successful no-op for an already formatted file is accepted. formatter.record(path) self.assertFalse((await formatter.finish())["changed"]) config = self.root / "treefmt.toml" config.rename(self.root / "shared.toml") config.symlink_to("shared.toml") self.fs.write(str(path), "lower again\n") self.assertTrue((await formatter.finish())["changed"]) self.assertEqual(path.read_text(), "context-formatted\n")
@unittest.skipUnless( shutil.which("nix") and shutil.which("treefmt") and shutil.which("git"), "standalone Nix smoke requires nix, treefmt and git", ) async def test_real_nix_fmt_treefmt_exact_path(self): nix = shutil.which("nix") treefmt = shutil.which("treefmt") assert nix is not None and treefmt is not None
async def command(*argv): child = await asyncio.create_subprocess_exec( *argv, cwd=self.root, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) output, _ = await asyncio.wait_for(child.communicate(), 30) self.assertEqual(child.returncode, 0, output.decode()) return output.decode().strip()
system = await command( "nix", "eval", "--impure", "--raw", "--expr", "builtins.currentSystem" ) treefmt_root = str(Path(treefmt).resolve().parent.parent) for root_flags in ([], ["--tree-root-file", "flake.nix"]): wrapper = ( f"#!{sys.executable}\n" "import os, sys\n" f"os.execv({treefmt_root + '/bin/treefmt'!r}, " f"[{treefmt_root + '/bin/treefmt'!r}, '--no-cache', " f"'--config-file', 'project.toml', *{root_flags!r}, *sys.argv[1:]])\n" ) builder = ( "import os, pathlib\n" "p = pathlib.Path(os.environ['out']) / 'bin' / 'formatter'\n" "p.parent.mkdir(parents=True)\n" f"p.write_text({wrapper!r})\n" "p.chmod(0o755)\n" ) (self.root / "flake.nix").write_text( "{ outputs = _: { formatter." + system + " = builtins.derivation { " 'name = "formatter-shadow-smoke"; system = ' + json.dumps(system) + "; " 'builder = "${builtins.storePath ' + json.dumps(str(Path(sys.executable).resolve().parent.parent)) + '}/bin/python3"; ' 'args = [ "-c" ' + json.dumps(builder) + " ]; " "treefmt = builtins.storePath " + json.dumps(treefmt_root) + "; " '} // { meta.mainProgram = "formatter"; }; }; }' ) (self.root / "project.toml").write_text( "[formatter.exact]\n" f"command = {json.dumps(sys.executable)}\n" 'options = ["-c", "import pathlib,sys; p=pathlib.Path(sys.argv[1]); p.write_text(p.read_text().upper())"]\n' 'includes = ["src/example.py"]\n' ) (self.root / "src").mkdir(exist_ok=True) path = self.write("src/example.py") await command("git", "init", "--quiet") await command("git", "add", "flake.nix", "project.toml", "src/example.py") self.config["nix"] = [ nix, "--offline", "--option", "pure-eval", "false", ] self.config["timeoutMs"] = 30000 result = await formatter.finish() self.assertTrue(result["ok"], result["error"]) self.assertEqual(result, {"ok": True, "changed": True, "error": ""}) self.assertEqual(path.read_text(), "LOWER\n")
async def test_project_snapshot_reused_and_publish_guards(self): (self.root / "treefmt.toml").write_text("") first = self.write() second = self.write("second.py") with patch.object( formatter, "_snapshot", wraps=formatter._snapshot ) as snapshot: self.assertTrue((await formatter.finish())["ok"]) self.assertEqual(snapshot.call_count, 1) self.assertEqual(first.read_text(), "LOWER\n") self.assertEqual(second.read_text(), "LOWER\n") self.fs.write(str(first), "lower again") self.set_fixture("concurrent_change", original=str(first)) result = await formatter.finish() self.assertFalse(result["ok"]) self.assertIn("concurrent", result["error"]) self.assertEqual(first.read_text(), "user change") self.assertIn(first, formatter._pending)
async def test_project_cannot_redirect_later_target_reset(self): (self.root / "treefmt.toml").write_text("") first = self.write() second = self.write("second.py")
async def redirect_later_target(argv, cwd, **kwargs): staged_second = Path(argv[-1]).parent / second.name staged_second.unlink() staged_second.symlink_to(second)
with patch.object(formatter, "_run", side_effect=redirect_later_target): result = await formatter.finish() self.assertFalse(result["ok"]) self.assertIn("symlink", result["error"]) self.assertEqual(first.read_text(), "lower\n") self.assertEqual(second.read_text(), "lower\n") self.assertIn(second, formatter._pending)
async def test_project_staged_symlink_is_never_published(self): (self.root / "treefmt.toml").write_text("") path = self.write()
async def replace_with_symlink(argv, cwd, **kwargs): staged = Path(argv[-1]) staged.unlink() staged.symlink_to(path)
with patch.object(formatter, "_run", side_effect=replace_with_symlink): result = await formatter.finish() self.assertFalse(result["ok"]) self.assertIn("symlink", result["error"]) self.assertEqual(path.read_text(), "lower\n") self.assertIn(path, formatter._pending)
async def test_project_snapshot_limits_and_config_symlink_escape(self): (self.root / "treefmt.toml").write_text("") path = self.write() with patch.object(formatter, "MAX_CONTEXT_FILES", 1): self.assertIn("context exceeds", (await formatter.finish())["error"]) self.assertEqual(path.read_text(), "lower\n") self.assertIn(path, formatter._pending) with tempfile.TemporaryDirectory() as outside: external = Path(outside) / "treefmt.toml" external.write_text("") (self.root / "treefmt.toml").unlink() (self.root / "treefmt.toml").symlink_to(external) self.assertIn("outside workspace", (await formatter.finish())["error"]) self.assertEqual(path.read_text(), "lower\n")
async def test_busy_skips_without_losing_pending(self): path = self.write() job = types.SimpleNamespace(service=False, exit_code=None) with patch.dict( sys.modules, {"albedo_plugins.run": types.SimpleNamespace(jobs={"job": job})}, ): self.assertTrue((await formatter.finish())["changed"]) self.assertNotIn(path, formatter._pending) job.service = True self.assertFalse((await formatter.finish())["changed"]) self.write() other = asyncio.create_task(asyncio.sleep(20)) try: with patch.dict( sys.modules, {"albedo_cells": types.SimpleNamespace(CELL_TASKS={"other": other})}, ): self.assertTrue((await formatter.finish())["changed"]) self.assertFalse(formatter._pending) finally: other.cancel() with self.assertRaises(asyncio.CancelledError): await other
async def test_nix_missing_formatter_falls_back_not_build_failure(self): path = self.write() (self.root / "flake.nix").write_text("") nix_control = self.root / "nix-state.json" diagnostic = ( "error: flake 'path:/workspace' does not provide attribute " "'packages.x86_64-linux.formatter', 'legacyPackages.x86_64-linux.formatter' " "or 'formatter.x86_64-linux'" ) self.config["nix"] = [sys.executable, str(self.script), str(nix_control)] self.set_fixture("failure", control=nix_control, diagnostic=diagnostic) result = await formatter.finish() self.assertTrue(result["ok"]) self.assertTrue(result["changed"]) self.assertEqual(path.read_text(), "LOWER\n") self.write(text="lower again") single = "error: flake 'path:/workspace' does not provide attribute 'formatter.x86_64-linux'" self.set_fixture("failure", control=nix_control, diagnostic=single) self.assertTrue((await formatter.finish())["ok"]) self.assertEqual(path.read_text(), "LOWER AGAIN") self.write(text="lower again") self.set_fixture( "failure", control=nix_control, diagnostic="error: builder failed" ) result = await formatter.finish() self.assertFalse(result["ok"]) self.assertEqual(path.read_text(), "lower again")
async def test_remote_setup_noop(self): with ( patch.dict(os.environ, {"ALBEDO_REMOTE_TARGET": "host"}), patch.object(formatter, "configure") as configure, ): self.assertEqual(await plugin.setup(types.SimpleNamespace()), {}) configure.assert_not_called()
async def test_total_budget(self): first = self.write() second = self.write("second.py") self.config["timeoutMs"] = 1000 self.set_fixture("slow") result = await formatter.finish() self.assertFalse(result["ok"]) self.assertTrue(result["changed"]) self.assertEqual(first.read_text(), "done") self.assertEqual(second.read_text(), "lower\n") self.assertIn(second, formatter._pending) self.assertFalse(list(self.root.glob("albedo-format-*")))
async def test_setup_empty_exports(self): async def host(name, args): self.assertEqual(name, "formatter.config") self.assertEqual(args, {}) return self.config
with patch.object(formatter, "configure") as configure: self.assertEqual(await plugin.setup(types.SimpleNamespace(host=host)), {}) configure.assert_called_once_with(self.config)
if __name__ == "__main__": unittest.main()