kstyagi
3e17e3e72f
- Added new API endpoints for tool catalog and installation management. - Updated usage documentation to include new tool installation features. - Implemented asynchronous tool installation with job tracking. - Introduced integration tests for tool installation in disposable VMs. - Enhanced dashboard UI to reflect tool installation status and usage instructions. - Improved error handling and logging for tool installation processes. - Added support for multiple Python versions and Conda environments. - Refactored tool management logic to ensure better state handling and job persistence.
163 lines
8.6 KiB
Python
163 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import hashlib
|
|
import io
|
|
import shlex
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from uvm import guest_tools
|
|
from uvm.errors import ValidationError
|
|
from uvm.tool_catalog import CATALOG, ToolSelection, all_selections, normalize_tools, recipe
|
|
from uvm.tools import guest_command
|
|
|
|
|
|
class ToolCatalogTests(unittest.TestCase):
|
|
def test_requested_catalog_and_every_guest_payload_fit_the_agent_protocol(self):
|
|
self.assertEqual({tool["id"] for tool in CATALOG},
|
|
{"git", "nodejs", "python", "jdk", "wget", "curl", "htop", "php", "conda", "micromamba"})
|
|
for arch in ("x86_64", "aarch64"):
|
|
recipes = [recipe(item, arch) for item in all_selections()]
|
|
command = guest_command({"action": "probe", "tools": recipes, "username": "worker", "architecture": arch})
|
|
self.assertLess(len(command.encode()), 65536)
|
|
argv = shlex.split(command)
|
|
self.assertEqual(argv[:3], ["/usr/bin/python3", "-I", "-c"])
|
|
compile(argv[3], "guest-tools", "exec")
|
|
ast.parse(argv[3], feature_version=(3, 10))
|
|
for item in recipes:
|
|
if "url" in item:
|
|
self.assertTrue(item["url"].startswith("https://"))
|
|
self.assertNotIn("/latest", item["url"])
|
|
self.assertRegex(item["sha256"], r"^[0-9a-f]{64}$")
|
|
self.assertEqual(len(all_selections()), 12)
|
|
|
|
def test_selection_validation_is_allowlisted_strict_and_deduplicated(self):
|
|
expected = (ToolSelection("git"), ToolSelection("python", "3.12"))
|
|
self.assertEqual(normalize_tools([{"id": "python", "version": "3.12"}, {"id": "git"}, {"id": "git"}]), expected)
|
|
for value in (None, "git", [], ["git"], [{"id": "git; id"}], [{"id": "python"}],
|
|
[{"id": "python", "version": "3.14"}], [{"id": "git", "version": "latest"}],
|
|
[{"id": "git", "command": "id"}], [{"id": ["git"]}], [{"id": "git"}] * 13):
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
normalize_tools(value)
|
|
self.assertEqual(normalize_tools([], allow_empty=True), ())
|
|
|
|
def test_versioned_python_commands_and_jdk_compiler_are_verified(self):
|
|
for version in ("3.11", "3.12", "3.13"):
|
|
item = recipe(ToolSelection("python", version), "aarch64")
|
|
self.assertEqual(set(item["links"]), {"python" + version})
|
|
self.assertIn("venv", item["checks"][0][-1])
|
|
self.assertEqual(item["checks"][1][:3], ["python" + version, "-m", "pip"])
|
|
jdk = recipe(ToolSelection("jdk"), "aarch64")
|
|
self.assertEqual(jdk["checks"], [["javac", "-version"], ["java", "-version"]])
|
|
self.assertIn("arm64", jdk["java_home"])
|
|
|
|
|
|
class GuestToolRecipeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = self.enterContext(tempfile.TemporaryDirectory())
|
|
self.root = Path(self.temporary)
|
|
|
|
def test_download_mismatch_removes_artifact_and_https_redirect_is_checked(self):
|
|
target = self.root / "download"
|
|
response = io.BytesIO(b"publisher payload")
|
|
response.geturl = lambda: "https://publisher.test/tool"
|
|
with patch.object(guest_tools.urllib.request, "urlopen", return_value=response):
|
|
with self.assertRaisesRegex(guest_tools.ToolError, "checksum"):
|
|
guest_tools.download("https://publisher.test/tool", target, "0" * 64)
|
|
self.assertFalse(target.exists())
|
|
response = io.BytesIO(b"publisher payload")
|
|
response.geturl = lambda: "https://publisher.test/tool"
|
|
with patch.object(guest_tools.urllib.request, "urlopen", return_value=response):
|
|
guest_tools.download("https://publisher.test/tool", target, hashlib.sha256(b"publisher payload").hexdigest())
|
|
self.assertEqual(target.read_bytes(), b"publisher payload")
|
|
with self.assertRaisesRegex(guest_tools.ToolError, "HTTPS"):
|
|
guest_tools.download("http://publisher.test/tool", target)
|
|
response = io.BytesIO(b"publisher payload")
|
|
response.geturl = lambda: "http://publisher.test/tool"
|
|
with patch.object(guest_tools.urllib.request, "urlopen", return_value=response):
|
|
with self.assertRaisesRegex(guest_tools.ToolError, "HTTPS"):
|
|
guest_tools.download("https://publisher.test/tool", target)
|
|
|
|
def test_archive_traversal_links_and_devices_cannot_escape_staging(self):
|
|
cases = [("../outside", tarfile.REGTYPE, ""), ("/outside", tarfile.REGTYPE, ""),
|
|
("link", tarfile.SYMTYPE, "../../outside"), ("link", tarfile.LNKTYPE, "../outside"),
|
|
("device", tarfile.CHRTYPE, "")]
|
|
for name, kind, link in cases:
|
|
with self.subTest(name=name, kind=kind):
|
|
archive = self.root / "archive.tar"
|
|
with tarfile.open(archive, "w") as stream:
|
|
item = tarfile.TarInfo(name)
|
|
item.type = kind
|
|
item.linkname = link
|
|
stream.addfile(item)
|
|
output = self.root / "output"
|
|
output.mkdir(exist_ok=True)
|
|
with self.assertRaises(guest_tools.ToolError):
|
|
guest_tools.extract_archive(archive, output)
|
|
self.assertFalse((self.root / "outside").exists())
|
|
|
|
def test_internal_archive_symlinks_are_supported(self):
|
|
archive = self.root / "archive.tar"
|
|
with tarfile.open(archive, "w") as stream:
|
|
item = tarfile.TarInfo("python/bin/python3.12")
|
|
item.size = 4
|
|
item.mode = 0o755
|
|
stream.addfile(item, io.BytesIO(b"test"))
|
|
link = tarfile.TarInfo("python/bin/python3")
|
|
link.type = tarfile.SYMTYPE
|
|
link.linkname = "python3.12"
|
|
stream.addfile(link)
|
|
output = self.root / "output"
|
|
output.mkdir()
|
|
guest_tools.extract_archive(archive, output)
|
|
self.assertEqual((output / "python/bin/python3").read_bytes(), b"test")
|
|
|
|
def test_probe_uses_guest_account_and_rejects_wrong_runtime_or_missing_companion(self):
|
|
node = recipe(ToolSelection("nodejs"), "x86_64")
|
|
account = SimpleNamespace(pw_dir="/home/worker", pw_uid=1001, pw_gid=1002)
|
|
with (patch.object(guest_tools.pwd, "getpwnam", return_value=account),
|
|
patch.object(guest_tools.os, "getgrouplist", return_value=[1002, 1003]),
|
|
patch.object(guest_tools.shutil, "which", return_value="/usr/local/bin/node"),
|
|
patch.object(guest_tools.subprocess, "run") as run):
|
|
run.return_value = SimpleNamespace(returncode=0, stdout=b"v22.10.0\n")
|
|
result = guest_tools.probe(node, "worker")
|
|
self.assertFalse(result["installed"])
|
|
self.assertEqual(result["detected_version"], "22.10.0")
|
|
self.assertEqual(run.call_args.kwargs["user"], 1001)
|
|
self.assertEqual(run.call_args.kwargs["env"]["HOME"], "/home/worker")
|
|
run.side_effect = [SimpleNamespace(returncode=0, stdout=b"v24.21.0\n"),
|
|
SimpleNamespace(returncode=1, stdout=b"broken npm")]
|
|
self.assertFalse(guest_tools.probe(node, "worker")["installed"])
|
|
|
|
def test_managed_files_preserve_edits_and_can_finish_interrupted_publication(self):
|
|
installer = guest_tools.Installer("root")
|
|
path = self.root / "profile"
|
|
# Isolated directory fixtures stand in for the root-owned guest hierarchy.
|
|
with (patch.object(guest_tools, "ROOT", self.root),
|
|
patch.object(guest_tools, "safe_directory"),
|
|
patch.object(guest_tools, "regular_file")):
|
|
installer.write_managed(path, b"initial")
|
|
installer.write_managed(path, b"updated")
|
|
self.assertEqual(path.read_bytes(), b"updated")
|
|
path.write_bytes(b"guest edit")
|
|
with self.assertRaisesRegex(guest_tools.ToolError, "customized"):
|
|
installer.write_managed(path, b"next")
|
|
self.assertEqual(path.read_bytes(), b"guest edit")
|
|
path.write_bytes(b"updated")
|
|
installer.remember(path, hashlib.sha256(b"next").hexdigest())
|
|
installer.write_managed(path, b"next")
|
|
self.assertEqual(path.read_bytes(), b"next")
|
|
|
|
def test_log_capture_retains_a_bounded_tail(self):
|
|
installer = guest_tools.Installer("root")
|
|
installer.append(b"a" * 20000)
|
|
installer.append(b"last failure")
|
|
self.assertEqual(len(installer.log), guest_tools.LOG_LIMIT)
|
|
self.assertTrue(installer.truncated)
|
|
self.assertTrue(installer.log.endswith(b"last failure"))
|