khodowany
+57
-68
@@ -9,80 +9,75 @@ import tempfile
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import modulus.clangd as clangd
|
||||
import modulus.cli as cli
|
||||
import modulus.config as config
|
||||
import modulus.export as export
|
||||
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 = cli.build_context(self.root, config.normalize({"package": {"name": "sample"}}))
|
||||
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.image = "sha256:" + "1" * 64
|
||||
|
||||
def test_defaults_and_custom_settings(self):
|
||||
self.assertEqual(clangd.settings(self.context)["executable"], "/usr/bin/clangd-19")
|
||||
self.assertIn("/usr/bin/g++-12", clangd.settings(self.context)["query_drivers"])
|
||||
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 = config.normalize({"package": {"name": "sample"}, "tooling": {"lsp": {"clangd": custom}}})
|
||||
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(config.config_error):
|
||||
config.normalize({"package": {"name": "sample"}, "tooling": {"lsp": {"clangd": invalid}}})
|
||||
|
||||
{"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": {clangd.image_label: clangd.recipe_digest(self.context)}}}
|
||||
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(clangd.current_image(self.context), self.image)
|
||||
self.assertIsNone(clangd.current_image(self.context, "arm64"))
|
||||
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(clangd.current_image(self.context))
|
||||
|
||||
self.assertIsNone(modulus.clangd.current_image(self.context))
|
||||
def test_cache_is_shared_by_image_and_header_settings_not_repo(self):
|
||||
cache = clangd.cache_directory(self.context, self.image)
|
||||
other = cli.build_context(self.root / "other", self.context.config)
|
||||
self.assertEqual(cache, clangd.cache_directory(other, self.image))
|
||||
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, clangd.cache_directory(self.context, self.image))
|
||||
self.assertEqual(cache, modulus.clangd.cache_directory(self.context, self.image))
|
||||
self.context.lsp_settings["clangd"]["header_roots"].append("/opt/sdk")
|
||||
self.assertNotEqual(cache, clangd.cache_directory(self.context, self.image))
|
||||
self.assertNotEqual(cache, clangd.cache_directory(other, "sha256:" + "2" * 64))
|
||||
|
||||
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 = clangd.launch_command(self.context, [], self.image, self.root / "cache", ["/usr/include"], "test-server")
|
||||
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", str(pathlib.Path.home()) + ":"):
|
||||
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,/usr/bin/aarch64-linux-gnu-g++", command)
|
||||
|
||||
def test_startup_failure_does_not_prepare_and_keeps_stdout_clean(self):
|
||||
output, errors = io.StringIO(), io.StringIO()
|
||||
with patch.object(clangd, "current_image", return_value=None), patch.object(clangd, "prepare") as prepare, \
|
||||
patch.object(cli, "ensure_build_image") as build, contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors):
|
||||
self.assertEqual(clangd.launch(self.context, []), 1)
|
||||
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(clangd.path_mappings(pathlib.Path("/cache"), ["/usr/include", "/opt/sdk"]),
|
||||
"--path-mappings=/cache/rootfs/usr/include=/usr/include,/cache/rootfs/opt/sdk=/opt/sdk")
|
||||
|
||||
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:
|
||||
@@ -92,69 +87,63 @@ class clangd_test(unittest.TestCase):
|
||||
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)):
|
||||
("usr/include/../bad", tarfile.REGTYPE), ("etc/passwd", tarfile.REGTYPE),
|
||||
("usr/include/link", tarfile.SYMTYPE)):
|
||||
with self.subTest(name=name), self.assertRaises(ValueError):
|
||||
clangd.extract_headers(self.archive(name, kind), self.root, ["/usr/include"])
|
||||
clangd.extract_headers(self.archive("usr/include/header.h"), self.root, ["/usr/include"])
|
||||
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": clangd.cache_version, "roots": ["/usr/include"]}))
|
||||
with patch.object(clangd, "cache_directory", return_value=cache), patch.object(clangd, "current_image", return_value=self.image), \
|
||||
patch.object(subprocess, "run") as run:
|
||||
self.assertEqual(clangd.prepare(self.context), 0)
|
||||
(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(clangd, "cache_directory", return_value=cache), patch.object(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(clangd.prepare(self.context), 1)
|
||||
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(clangd.run_server(["docker"], "test-server"), 0)
|
||||
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):
|
||||
export.sync_nvim_file(self.root, {})
|
||||
modulus.export.sync_nvim_file(self.root, {})
|
||||
path = self.root / ".nvim.lua"
|
||||
before = path.stat().st_mtime_ns
|
||||
export.sync_nvim_file(self.root, {})
|
||||
modulus.export.sync_nvim_file(self.root, {})
|
||||
self.assertEqual(before, path.stat().st_mtime_ns)
|
||||
export.sync_nvim_file(self.root, {"enabled": False})
|
||||
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()):
|
||||
export.sync_nvim_file(self.root, {})
|
||||
export.sync_nvim_file(self.root, {"enabled": False})
|
||||
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(cli, "in_docker", return_value=False), patch.object(cli, "docker_run", return_value=0) as run, \
|
||||
patch.object(clangd, "prepare", return_value=0) as prepare:
|
||||
self.assertEqual(cli.lsp(self.context, []), 0)
|
||||
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(cli.lsp(self.context, []), 1)
|
||||
self.assertEqual(modulus.cli.lsp(self.context, []), 1)
|
||||
prepare.assert_not_called()
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user