import contextlib import io import json import os import pathlib import subprocess import tarfile import tempfile import unittest from unittest.mock import MagicMock, patch import modulus.clangd import modulus.cli import modulus.config import modulus.export class clangd_test(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = pathlib.Path(temporary.name) self.context = modulus.cli.build_context(self.root, modulus.config.normalize({"package": {"name": "sample"}})) self.context.docker_marker = self.root / "not-in-container" self.context.dockerfile.parent.mkdir(parents=True) self.context.dockerfile.write_text("FROM sample\n") self.context.docker_inputs[0].write_text("autopep8==2.3.2\n") self.image = "sha256:" + "1" * 64 def test_defaults_and_custom_settings(self): self.assertEqual(modulus.clangd.settings(self.context)["executable"], "/usr/bin/clangd-19") self.assertIn("/usr/bin/g++-12", modulus.clangd.settings(self.context)["query_drivers"]) custom = {"executable": "/opt/bin/clangd", "arguments": [], "query_drivers": ["/opt/bin/g++"], "header_roots": ["/opt/sdk"]} result = modulus.config.normalize({"package": {"name": "sample"}, "tooling": {"lsp": {"clangd": custom}}}) self.assertEqual(result["tooling"]["lsp"]["clangd"], custom) def test_config_rejects_unsafe_or_unknown_settings(self): for invalid in ([], {"unknown": True}, {"executable": "clangd"}, {"arguments": "--background-index"}, {"query_drivers": ["/usr/bin/*"]}, {"header_roots": ["/"]}, {"header_roots": ["/usr/../etc"]}, {"query_drivers": ["/bin/g++,/bin/gcc"]}, {"arguments": ["--query-driver=*"]}, {"arguments": ["--path-mappings=/=/"]}): with self.subTest(invalid=invalid), self.assertRaises(modulus.config.config_error): modulus.config.normalize({"package": {"name": "sample"}, "tooling": {"lsp": {"clangd": invalid}}}) def test_image_fingerprint_and_architecture(self): image = {"Id": self.image, "Architecture": "amd64", "Config": {"Labels": {modulus.clangd.image_label: modulus.clangd.recipe_digest(self.context)}}} with patch.object(subprocess, "run", return_value=subprocess.CompletedProcess([], 0, json.dumps([image]))): self.assertEqual(modulus.clangd.current_image(self.context), self.image) self.assertIsNone(modulus.clangd.current_image(self.context, "arm64")) self.context.dockerfile.write_text("FROM changed\n") self.assertIsNone(modulus.clangd.current_image(self.context)) def test_cache_is_shared_by_image_and_header_settings_not_repo(self): cache = modulus.clangd.cache_directory(self.context, self.image) other = modulus.cli.build_context(self.root / "other", self.context.config) self.assertEqual(cache, modulus.clangd.cache_directory(other, self.image)) self.context.lsp_settings["clangd"]["arguments"].append("--log=error") self.assertEqual(cache, modulus.clangd.cache_directory(self.context, self.image)) self.context.lsp_settings["clangd"]["header_roots"].append("/opt/sdk") self.assertNotEqual(cache, modulus.clangd.cache_directory(self.context, self.image)) self.assertNotEqual(cache, modulus.clangd.cache_directory(other, "sha256:" + "2" * 64)) def test_stdio_launch_has_no_tty_socket_credentials_or_host_home(self): command = modulus.clangd.launch_command(self.context, [], self.image, self.root / "cache", ["/usr/include"], "test-server") self.assertIn("--interactive", command) self.assertIn("--read-only", command) self.assertIn(f"{self.root}:{self.root}:ro", command) self.assertIn(f"{self.context.build_root}:{self.context.build_root}:rw", command) self.assertIn("--network", command) self.assertNotIn("-t", command) for unwanted in ("--tty", "--volumes-from", "CONAN_TOKEN", "/var/run/docker.sock"): self.assertNotIn(unwanted, " ".join(command)) volumes = [command[index + 1] for index, argument in enumerate(command) if argument == "--volume"] self.assertFalse(any(volume.startswith(str(pathlib.Path.home()) + ":") for volume in volumes)) self.assertIn("--query-driver=/usr/bin/gcc-12,/usr/bin/g++-12,/usr/bin/aarch64-linux-gnu-gcc-12,/usr/bin/aarch64-linux-gnu-g++-12", command) def test_startup_failure_does_not_prepare_and_keeps_stdout_clean(self): output, errors = io.StringIO(), io.StringIO() with patch.object(modulus.clangd, "current_image", return_value=None), patch.object(modulus.clangd, "prepare") as prepare, \ patch.object(modulus.cli, "ensure_build_image") as build, contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): self.assertEqual(modulus.clangd.launch(self.context, []), 1) prepare.assert_not_called() build.assert_not_called() self.assertEqual(output.getvalue(), "") self.assertIn("configure", errors.getvalue()) def test_path_mapping_uses_client_to_server_direction_and_specificity(self): self.assertEqual(modulus.clangd.path_mappings(pathlib.Path("/cache"), ["/usr/include", "/opt/sdk"]), "--path-mappings=/cache/rootfs/usr/include=/usr/include,/cache/rootfs/opt/sdk=/opt/sdk") def archive(self, name, kind=tarfile.REGTYPE): stream = io.BytesIO() with tarfile.open(fileobj=stream, mode="w") as archive: member = tarfile.TarInfo(name) member.type = kind member.linkname = "/etc/passwd" if kind == tarfile.SYMTYPE else "" archive.addfile(member) stream.seek(0) return stream def test_header_extraction_rejects_escape_and_links(self): for name, kind in (("../bad", tarfile.REGTYPE), ("/usr/include/bad", tarfile.REGTYPE), ("usr/include/../bad", tarfile.REGTYPE), ("etc/passwd", tarfile.REGTYPE), ("usr/include/link", tarfile.SYMTYPE)): with self.subTest(name=name), self.assertRaises(ValueError): modulus.clangd.extract_headers(self.archive(name, kind), self.root, ["/usr/include"]) modulus.clangd.extract_headers(self.archive("usr/include/header.h"), self.root, ["/usr/include"]) self.assertEqual((self.root / "usr/include/header.h").stat().st_mode & 0o777, 0o444) def test_completed_cache_never_invokes_container_again(self): cache = self.root / "cache" (cache / "rootfs/usr/include").mkdir(parents=True) (cache / "manifest.json").write_text(json.dumps({"version": modulus.clangd.cache_version, "roots": ["/usr/include"]})) with patch.object(modulus.clangd, "cache_directory", return_value=cache), patch.object(modulus.clangd, "current_image", return_value=self.image), \ patch.object(subprocess, "run") as run: self.assertEqual(modulus.clangd.prepare(self.context), 0) run.assert_not_called() def test_failed_extraction_never_publishes_partial_cache(self): cache = self.root / "image/cache" process = MagicMock() process.__enter__.return_value = process process.stdout = self.archive("../escape") with patch.object(modulus.clangd, "cache_directory", return_value=cache), patch.object(modulus.clangd, "current_image", return_value=self.image), \ patch.object(subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '["/usr/include"]')), \ patch.object(subprocess, "Popen", return_value=process), contextlib.redirect_stderr(io.StringIO()): self.assertEqual(modulus.clangd.prepare(self.context), 1) self.assertFalse(cache.exists()) self.assertEqual(list(cache.parent.glob(".headers-*")), []) def test_cleanup_always_waits_for_child_even_if_docker_removal_times_out(self): process = MagicMock() process.wait.side_effect = [0, subprocess.TimeoutExpired("docker", 5), 0] with patch.object(subprocess, "Popen", return_value=process), \ patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired("docker rm", 15)), \ contextlib.redirect_stderr(io.StringIO()): self.assertEqual(modulus.clangd.run_server(["docker"], "test-server"), 0) process.kill.assert_called_once() self.assertEqual(process.wait.call_count, 3) def test_nvim_generation_is_stable_opt_out_and_respects_ownership(self): modulus.export.sync_nvim_file(self.root, {}) path = self.root / ".nvim.lua" before = path.stat().st_mtime_ns modulus.export.sync_nvim_file(self.root, {}) self.assertEqual(before, path.stat().st_mtime_ns) modulus.export.sync_nvim_file(self.root, {"enabled": False}) self.assertFalse(path.exists()) path.write_text("-- user settings\n") with contextlib.redirect_stderr(io.StringIO()): modulus.export.sync_nvim_file(self.root, {}) modulus.export.sync_nvim_file(self.root, {"enabled": False}) self.assertEqual(path.read_text(), "-- user settings\n") def test_lsp_prepares_after_successful_database_refresh_only(self): with patch.object(modulus.cli, "in_docker", return_value=False), patch.object(modulus.cli, "docker_run", return_value=0) as run, \ patch.object(modulus.clangd, "prepare", return_value=0) as prepare: self.assertEqual(modulus.cli.lsp(self.context, []), 0) prepare.assert_called_once_with(self.context) prepare.reset_mock() run.return_value = 1 self.assertEqual(modulus.cli.lsp(self.context, []), 1) prepare.assert_not_called() if __name__ == "__main__": unittest.main()