refactor: separate engine, collector, mutator, and fleet into independent subpackages
- decnet/engine/ — container lifecycle (deploy, teardown, status); _kill_api removed - decnet/collector/ — Docker log streaming (moved from web/collector.py) - decnet/mutator/ — mutation engine (no longer imports from cli or duplicates deployer code) - decnet/fleet.py — shared decky-building logic extracted from cli.py Cross-contamination eliminated: - web router no longer imports from decnet.cli - mutator no longer imports from decnet.cli - cli no longer imports from decnet.web - _kill_api() moved to cli (process management, not engine concern) - _compose_with_retry duplicate removed from mutator
This commit is contained in:
@@ -3,7 +3,7 @@
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
from decnet.web.collector import parse_rfc5424, is_service_container, is_service_event
|
||||
from decnet.collector import parse_rfc5424, is_service_container, is_service_event
|
||||
|
||||
_KNOWN_NAMES = {"omega-decky-http", "omega-decky-smtp", "relay-decky-ftp"}
|
||||
|
||||
@@ -91,42 +91,42 @@ class TestParseRfc5424:
|
||||
|
||||
class TestIsServiceContainer:
|
||||
def test_known_container_returns_true(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_container(_make_container("omega-decky-http")) is True
|
||||
assert is_service_container(_make_container("omega-decky-smtp")) is True
|
||||
assert is_service_container(_make_container("relay-decky-ftp")) is True
|
||||
|
||||
def test_base_container_returns_false(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_container(_make_container("omega-decky")) is False
|
||||
|
||||
def test_unrelated_container_returns_false(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_container(_make_container("nginx")) is False
|
||||
|
||||
def test_strips_leading_slash(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_container(_make_container("/omega-decky-http")) is True
|
||||
assert is_service_container(_make_container("/omega-decky")) is False
|
||||
|
||||
def test_no_state_returns_false(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=set()):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=set()):
|
||||
assert is_service_container(_make_container("omega-decky-http")) is False
|
||||
|
||||
|
||||
class TestIsServiceEvent:
|
||||
def test_known_service_event_returns_true(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_event({"name": "omega-decky-smtp"}) is True
|
||||
|
||||
def test_base_event_returns_false(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_event({"name": "omega-decky"}) is False
|
||||
|
||||
def test_unrelated_event_returns_false(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=_KNOWN_NAMES):
|
||||
assert is_service_event({"name": "nginx"}) is False
|
||||
|
||||
def test_no_state_returns_false(self):
|
||||
with patch("decnet.web.collector._load_service_container_names", return_value=set()):
|
||||
with patch("decnet.collector.worker._load_service_container_names", return_value=set()):
|
||||
assert is_service_event({"name": "omega-decky-smtp"}) is False
|
||||
|
||||
Reference in New Issue
Block a user