67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
"""Unit: smolmachines active-agent enumeration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from bot_bottle.backend.smolmachines import enumerate as _enumerate
|
|
|
|
|
|
def _completed(
|
|
stdout: str,
|
|
returncode: int = 0,
|
|
) -> subprocess.CompletedProcess: # type: ignore[type-arg]
|
|
return subprocess.CompletedProcess(
|
|
args=[],
|
|
returncode=returncode,
|
|
stdout=stdout,
|
|
stderr="",
|
|
)
|
|
|
|
|
|
class TestEnumerateActive(unittest.TestCase):
|
|
def test_skips_sidecar_vms(self):
|
|
machines = [
|
|
{"name": "bot-bottle-demo-abc12", "state": "running"},
|
|
{"name": "bot-bottle-sidecars-demo-abc12", "state": "running"},
|
|
{"name": "bot-bottle-stopped-xyz", "state": "stopped"},
|
|
{"name": "other", "state": "running"},
|
|
]
|
|
|
|
class _Metadata:
|
|
agent_name = "demo"
|
|
started_at = "2026-07-09T00:00:00Z"
|
|
label = "Demo"
|
|
color = "green"
|
|
|
|
with patch(
|
|
"bot_bottle.backend.smolmachines.enumerate.subprocess.run",
|
|
return_value=_completed(json.dumps(machines)),
|
|
), patch(
|
|
"bot_bottle.backend.smolmachines.enumerate._query_bundle_services",
|
|
return_value={"demo-abc12": ("egress",)},
|
|
), patch(
|
|
"bot_bottle.backend.smolmachines.enumerate.read_metadata",
|
|
return_value=_Metadata(),
|
|
) as read_metadata:
|
|
active = _enumerate.enumerate_active()
|
|
|
|
self.assertEqual(1, len(active))
|
|
self.assertEqual("demo-abc12", active[0].slug)
|
|
self.assertEqual(("egress",), active[0].services)
|
|
read_metadata.assert_called_once_with("demo-abc12")
|
|
|
|
def test_invalid_smolvm_output_returns_empty(self):
|
|
with patch(
|
|
"bot_bottle.backend.smolmachines.enumerate.subprocess.run",
|
|
return_value=_completed("not json"),
|
|
):
|
|
self.assertEqual([], _enumerate.enumerate_active())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|