khodowany
162 lines
8.6 KiB
Python
162 lines
8.6 KiB
Python
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 as clangd
|
|
import modulus.cli as cli
|
|
import modulus.config as config
|
|
import modulus.export as 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.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"])
|
|
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}}})
|
|
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}}})
|
|
|
|
def test_image_fingerprint_and_architecture(self):
|
|
image = {"Id": self.image, "Architecture": "amd64", "Config": {"Labels": {clangd.image_label: 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.context.dockerfile.write_text("FROM changed\n")
|
|
self.assertIsNone(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))
|
|
self.context.lsp_settings["clangd"]["arguments"].append("--log=error")
|
|
self.assertEqual(cache, 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))
|
|
|
|
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")
|
|
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()) + ":"):
|
|
self.assertNotIn(unwanted, " ".join(command))
|
|
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)
|
|
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")
|
|
|
|
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):
|
|
clangd.extract_headers(self.archive(name, kind), self.root, ["/usr/include"])
|
|
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)
|
|
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)
|
|
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)
|
|
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, {})
|
|
path = self.root / ".nvim.lua"
|
|
before = path.stat().st_mtime_ns
|
|
export.sync_nvim_file(self.root, {})
|
|
self.assertEqual(before, path.stat().st_mtime_ns)
|
|
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})
|
|
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)
|
|
prepare.assert_called_once_with(self.context)
|
|
prepare.reset_mock()
|
|
run.return_value = 1
|
|
self.assertEqual(cli.lsp(self.context, []), 1)
|
|
prepare.assert_not_called()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|