Spaces:
Running
Running
File size: 6,908 Bytes
6337ffb 4671695 6337ffb 7ea15a2 6337ffb 27cb6df 6337ffb 7ea15a2 6337ffb fc5d520 6337ffb 3052732 6337ffb fc5d520 6337ffb 7ea15a2 6337ffb f407578 211a1f0 f407578 a663402 085f660 a663402 211a1f0 085f660 211a1f0 3052732 bf947b8 3052732 4671695 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | from __future__ import annotations
import io
import unittest
from contextlib import redirect_stderr, redirect_stdout
from unittest.mock import patch
from lumi_eggcracker.cli import main
class CliTests(unittest.TestCase):
def test_version(self) -> None:
output = io.StringIO()
with redirect_stdout(output):
self.assertEqual(0, main(["version"]))
self.assertEqual("1.0.10", output.getvalue().strip())
def test_public_help_has_only_supported_commands(self) -> None:
from lumi_eggcracker.cli import _parser
help_text = _parser().format_help()
for command in (
"start",
"kill",
"status",
"list",
"approve",
"revoke",
"approvals",
"detections",
"doctor",
"support-bundle",
"validate-support-bundle",
"version",
):
self.assertIn(command, help_text)
self.assertIn("exec-policy", help_text)
self.assertNotIn("_supervisor", help_text)
self.assertNotIn("network" + "-deny", help_text)
def test_validate_support_bundle_dispatches_before_connected_commands(self) -> None:
with (
patch("lumi_eggcracker.cli.validate_support_bundle_main", return_value=0) as validate,
patch("lumi_eggcracker.cli.request") as request,
patch("lumi_eggcracker.cli.supervisor_main") as supervisor,
patch("lumi_eggcracker.cli.gate_main") as gate,
patch("lumi_eggcracker.cli.watchdog_main") as watchdog,
):
self.assertEqual(
0,
main(["validate-support-bundle", "private-name-must-not-be-echoed.json"]),
)
validate.assert_called_once_with(["private-name-must-not-be-echoed.json"])
request.assert_not_called()
supervisor.assert_not_called()
gate.assert_not_called()
watchdog.assert_not_called()
def test_internal_supervisor_dispatch_remains_available(self) -> None:
with patch("lumi_eggcracker.cli.supervisor_main", return_value=7) as supervisor:
self.assertEqual(main(["_supervisor", "--policy", "/tmp/policy.json"]), 7)
supervisor.assert_called_once_with(["--policy", "/tmp/policy.json"])
def test_approve_forwards_only_exact_command_arguments(self) -> None:
with patch("lumi_eggcracker.cli.os.geteuid", return_value=0, create=True), patch("lumi_eggcracker.cli.request", return_value={"result": "APPROVED"}) as request:
self.assertEqual(0, main(["approve", "--name", "qwen", "--uid", "1001", "--", "/opt/llama-cli", "-m", "/models/qwen.gguf"]))
request.assert_called_once_with(
"approve",
name="qwen",
uid=1001,
max_pids=64,
max_memory_mib=2048,
cpu_quota_percent=400,
allow_interface_discovery=False,
argv=["/opt/llama-cli", "-m", "/models/qwen.gguf"],
)
def test_approve_can_grant_interface_discovery(self) -> None:
with patch("lumi_eggcracker.cli.os.geteuid", return_value=0, create=True), patch("lumi_eggcracker.cli.request", return_value={"result": "APPROVED"}) as request:
self.assertEqual(0, main(["approve", "--name", "qwen", "--uid", "1001", "--allow-interface-discovery", "--", "/opt/llama-cli", "-m", "/models/qwen.gguf"]))
self.assertTrue(request.call_args.kwargs["allow_interface_discovery"])
def test_start_includes_resource_limits(self) -> None:
with patch("lumi_eggcracker.cli.request", return_value={"result": "STARTED"}) as request:
self.assertEqual(0, main(["start", "--name", "demo", "--max-pids", "8", "--", "/bin/sleep", "1"]))
request.assert_called_once_with(
"start", name="demo", max_pids=8, max_memory_mib=2048,
cpu_quota_percent=400, argv=["/bin/sleep", "1"],
)
def test_start_forwards_selected_execution_policy(self) -> None:
with patch("lumi_eggcracker.cli.request", return_value={"result": "STARTED"}) as request:
self.assertEqual(0, main(["start", "--name", "demo", "--exec-policy", "a" * 24, "--max-pids", "8", "--", "/bin/sleep", "1"]))
request.assert_called_once_with(
"start", name="demo", max_pids=8, max_memory_mib=2048,
cpu_quota_percent=400, argv=["/bin/sleep", "1"], exec_policy="a" * 24,
)
def test_start_can_require_exact_approval_at_admission_time(self) -> None:
with patch("lumi_eggcracker.cli.request", return_value={"result": "STARTED"}) as request:
self.assertEqual(
0,
main(
[
"start",
"--name",
"demo",
"--max-pids",
"8",
"--require-approval",
"--",
"/bin/sleep",
"1",
]
),
)
request.assert_called_once_with(
"start",
name="demo",
max_pids=8,
max_memory_mib=2048,
cpu_quota_percent=400,
argv=["/bin/sleep", "1"],
require_approval=True,
)
def test_execution_policy_create_requires_root(self) -> None:
with patch("lumi_eggcracker.cli.os.geteuid", return_value=1001, create=True):
self.assertEqual(4, main(["exec-policy", "create", "--name", "demo", "--", "/bin/sh"]))
def test_duplicate_identifier_options_are_rejected_before_request(self) -> None:
policy_a = "a" * 24
policy_b = "b" * 24
cases = (
["start", "--name", "first", "--name", "second", "--max-pids", "8", "--", "/bin/true"],
["start", "--name", "demo", "--exec-policy", policy_a, "--exec-policy", policy_b, "--max-pids", "8", "--", "/bin/true"],
["kill", "--name", "first", "--name", "second", "--receipt", "/tmp/receipt.json"],
["status", "--name", "first", "--name", "second"],
["approve", "--name", "first", "--name", "second", "--uid", "1001", "--", "/bin/true"],
["revoke", "--name", "first", "--name", "second"],
["exec-policy", "create", "--name", "first", "--name", "second", "--", "/bin/true"],
["exec-policy", "revoke", "--policy-id", policy_a, "--policy-id", policy_b],
)
for values in cases:
with self.subTest(values=values), patch(
"lumi_eggcracker.cli.request"
) as request, redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit) as raised:
main(values)
self.assertEqual(2, raised.exception.code)
request.assert_not_called()
|