From 4ae6f4f23db5cc1e891efae72a33426001e76643 Mon Sep 17 00:00:00 2001 From: anti Date: Thu, 9 Apr 2026 12:55:52 -0400 Subject: [PATCH] =?UTF-8?q?test:=20expand=20coverage=2064%=E2=86=9276%;=20?= =?UTF-8?q?add=20BUGS.md=20for=20Gemini=20migration=20issues?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- BUGS.md | 37 ++++++ tests/test_config.py | 167 ++++++++++++++++++++++++++ tests/test_custom_service.py | 65 ++++++++++ tests/test_logging_forwarder.py | 62 ++++++++++ tests/test_mutator.py | 207 ++++++++++++++++++++++++++++++++ tests/test_network.py | 157 ++++++++++++++++++++++++ tests/test_services.py | 22 ++++ 7 files changed, 717 insertions(+) create mode 100644 BUGS.md create mode 100644 tests/test_config.py create mode 100644 tests/test_custom_service.py create mode 100644 tests/test_logging_forwarder.py create mode 100644 tests/test_mutator.py diff --git a/BUGS.md b/BUGS.md new file mode 100644 index 0000000..63ff8ee --- /dev/null +++ b/BUGS.md @@ -0,0 +1,37 @@ +# BUGS + +Active bugs detected during development. Do not fix until noted otherwise. + +--- + +## BUG-001 — Split-brain model imports across router files (Gemini SQLModel migration) + +**Detected:** 2026-04-09 +**Status:** Open — do not fix, migration in progress + +**Symptom:** `from decnet.web.api import app` fails with `ModuleNotFoundError: No module named 'decnet.web.models'` + +**Root cause:** Gemini's SQLModel migration is partially complete. Models were moved to `decnet/web/db/models.py`, but three router files were not updated and still import from the old `decnet.web.models` path: + +| File | Stale import | +|------|--------------| +| `decnet/web/router/auth/api_login.py:12` | `from decnet.web.models import LoginRequest, Token` | +| `decnet/web/router/auth/api_change_pass.py:7` | `from decnet.web.models import ChangePasswordRequest` | +| `decnet/web/router/stats/api_get_stats.py:6` | `from decnet.web.models import StatsResponse` | + +**Fix:** Update those three files to import from `decnet.web.db.models` (consistent with the other router files already migrated). + +**Impact:** All `tests/api/` tests fail to collect. Web server cannot start. + +--- + +## BUG-002 — `decnet/web/db/sqlite/repository.py` depends on `sqlalchemy` directly + +**Detected:** 2026-04-09 +**Status:** Resolved (dependency installed via `pip install -e ".[dev]"`) + +**Symptom:** `ModuleNotFoundError: No module named 'sqlalchemy'` before `sqlmodel` was installed. + +**Root cause:** `sqlmodel>=0.0.16` was added to `pyproject.toml` but `pip install -e .` had not been re-run in the dev environment. + +**Fix:** Run `pip install -e ".[dev]"`. Already applied. diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..0cfa86f --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,167 @@ +""" +Tests for decnet.config — Pydantic models, save/load/clear state. +Covers the uncovered lines: validators, save_state, load_state, clear_state. +""" +import json +import pytest +from pathlib import Path +from unittest.mock import patch + +import decnet.config as config_module +from decnet.config import ( + DeckyConfig, + DecnetConfig, + save_state, + load_state, + clear_state, +) + + +# --------------------------------------------------------------------------- +# DeckyConfig validator +# --------------------------------------------------------------------------- + +class TestDeckyConfig: + def _base(self, **kwargs): + defaults = dict( + name="decky-01", ip="192.168.1.10", services=["ssh"], + distro="debian", base_image="debian", hostname="host-01", + ) + defaults.update(kwargs) + return defaults + + def test_valid_decky(self): + d = DeckyConfig(**self._base()) + assert d.name == "decky-01" + + def test_empty_services_raises(self): + with pytest.raises(Exception, match="at least one service"): + DeckyConfig(**self._base(services=[])) + + def test_multiple_services_ok(self): + d = DeckyConfig(**self._base(services=["ssh", "smb", "rdp"])) + assert len(d.services) == 3 + + +# --------------------------------------------------------------------------- +# DecnetConfig validator +# --------------------------------------------------------------------------- + +class TestDecnetConfig: + def _base_decky(self): + return DeckyConfig( + name="d", ip="10.0.0.2", services=["ssh"], + distro="debian", base_image="debian", hostname="h", + ) + + def test_valid_config(self): + cfg = DecnetConfig( + mode="unihost", interface="eth0", + subnet="10.0.0.0/24", gateway="10.0.0.1", + deckies=[self._base_decky()], + ) + assert cfg.mode == "unihost" + + def test_valid_log_target(self): + cfg = DecnetConfig( + mode="unihost", interface="eth0", + subnet="10.0.0.0/24", gateway="10.0.0.1", + deckies=[self._base_decky()], + log_target="192.168.1.5:5140", + ) + assert cfg.log_target == "192.168.1.5:5140" + + def test_none_log_target_ok(self): + cfg = DecnetConfig( + mode="unihost", interface="eth0", + subnet="10.0.0.0/24", gateway="10.0.0.1", + deckies=[self._base_decky()], + log_target=None, + ) + assert cfg.log_target is None + + def test_invalid_log_target_no_port(self): + with pytest.raises(Exception): + DecnetConfig( + mode="unihost", interface="eth0", + subnet="10.0.0.0/24", gateway="10.0.0.1", + deckies=[self._base_decky()], + log_target="192.168.1.5", + ) + + def test_invalid_log_target_non_digit_port(self): + with pytest.raises(Exception): + DecnetConfig( + mode="unihost", interface="eth0", + subnet="10.0.0.0/24", gateway="10.0.0.1", + deckies=[self._base_decky()], + log_target="192.168.1.5:abc", + ) + + +# --------------------------------------------------------------------------- +# save_state / load_state / clear_state +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def patch_state_file(tmp_path, monkeypatch): + monkeypatch.setattr(config_module, "STATE_FILE", tmp_path / "decnet-state.json") + + +def _sample_config(): + return DecnetConfig( + mode="unihost", interface="eth0", + subnet="192.168.1.0/24", gateway="192.168.1.1", + deckies=[ + DeckyConfig( + name="decky-01", ip="192.168.1.10", services=["ssh"], + distro="debian", base_image="debian", hostname="host-01", + ) + ], + log_target="10.0.0.1:5140", + ) + + +def test_save_and_load_state(tmp_path): + cfg = _sample_config() + compose = tmp_path / "docker-compose.yml" + save_state(cfg, compose) + + result = load_state() + assert result is not None + loaded_cfg, loaded_compose = result + assert loaded_cfg.mode == "unihost" + assert loaded_cfg.deckies[0].name == "decky-01" + assert loaded_cfg.log_target == "10.0.0.1:5140" + assert loaded_compose == compose + + +def test_load_state_returns_none_when_missing(tmp_path, monkeypatch): + monkeypatch.setattr(config_module, "STATE_FILE", tmp_path / "nonexistent.json") + assert load_state() is None + + +def test_clear_state(tmp_path): + cfg = _sample_config() + save_state(cfg, tmp_path / "compose.yml") + assert config_module.STATE_FILE.exists() + + clear_state() + assert not config_module.STATE_FILE.exists() + + +def test_clear_state_noop_when_missing(tmp_path, monkeypatch): + monkeypatch.setattr(config_module, "STATE_FILE", tmp_path / "nonexistent.json") + clear_state() # should not raise + + +def test_state_roundtrip_preserves_all_fields(tmp_path): + cfg = _sample_config() + cfg.deckies[0].archetype = "workstation" + cfg.deckies[0].mutate_interval = 45 + compose = tmp_path / "compose.yml" + save_state(cfg, compose) + + loaded_cfg, _ = load_state() + assert loaded_cfg.deckies[0].archetype == "workstation" + assert loaded_cfg.deckies[0].mutate_interval == 45 diff --git a/tests/test_custom_service.py b/tests/test_custom_service.py new file mode 100644 index 0000000..7f9a223 --- /dev/null +++ b/tests/test_custom_service.py @@ -0,0 +1,65 @@ +""" +Tests for decnet.custom_service — BYOS (bring-your-own-service) support. +""" +import pytest +from decnet.custom_service import CustomService + + +class TestCustomServiceComposeFragment: + def _svc(self, name="my-tool", image="myrepo/mytool:latest", + exec_cmd="", ports=None): + return CustomService(name=name, image=image, + exec_cmd=exec_cmd, ports=ports) + + def test_basic_fragment_structure(self): + svc = self._svc() + frag = svc.compose_fragment("decky-01") + assert frag["image"] == "myrepo/mytool:latest" + assert frag["container_name"] == "decky-01-my-tool" + assert frag["restart"] == "unless-stopped" + assert frag["environment"]["NODE_NAME"] == "decky-01" + + def test_underscores_in_name_become_dashes(self): + svc = self._svc(name="my_custom_tool") + frag = svc.compose_fragment("decky-01") + assert frag["container_name"] == "decky-01-my-custom-tool" + + def test_exec_cmd_is_split_into_list(self): + svc = self._svc(exec_cmd="/usr/bin/server --port 8080") + frag = svc.compose_fragment("decky-01") + assert frag["command"] == ["/usr/bin/server", "--port", "8080"] + + def test_empty_exec_cmd_omits_command_key(self): + svc = self._svc(exec_cmd="") + frag = svc.compose_fragment("decky-01") + assert "command" not in frag + + def test_log_target_injected_into_environment(self): + svc = self._svc() + frag = svc.compose_fragment("decky-01", log_target="10.0.0.5:5140") + assert frag["environment"]["LOG_TARGET"] == "10.0.0.5:5140" + + def test_no_log_target_omits_key(self): + svc = self._svc() + frag = svc.compose_fragment("decky-01", log_target=None) + assert "LOG_TARGET" not in frag["environment"] + + def test_service_cfg_is_accepted_without_error(self): + svc = self._svc() + # service_cfg is accepted but not used by CustomService + frag = svc.compose_fragment("decky-01", service_cfg={"key": "val"}) + assert frag is not None + + def test_ports_stored_on_instance(self): + svc = CustomService("tool", "img", "", ports=[8080, 9090]) + assert svc.ports == [8080, 9090] + + def test_no_ports_defaults_to_empty_list(self): + svc = CustomService("tool", "img", "") + assert svc.ports == [] + + +class TestCustomServiceDockerfileContext: + def test_returns_none(self): + svc = CustomService("tool", "img", "cmd") + assert svc.dockerfile_context() is None diff --git a/tests/test_logging_forwarder.py b/tests/test_logging_forwarder.py new file mode 100644 index 0000000..d345936 --- /dev/null +++ b/tests/test_logging_forwarder.py @@ -0,0 +1,62 @@ +""" +Tests for decnet.logging.forwarder — parse_log_target, probe_log_target. +""" +import socket +from unittest.mock import MagicMock, patch + +import pytest + +from decnet.logging.forwarder import parse_log_target, probe_log_target + + +class TestParseLogTarget: + def test_valid_ip_port(self): + host, port = parse_log_target("192.168.1.5:5140") + assert host == "192.168.1.5" + assert port == 5140 + + def test_valid_hostname_port(self): + host, port = parse_log_target("logstash.internal:9600") + assert host == "logstash.internal" + assert port == 9600 + + def test_no_colon_raises(self): + with pytest.raises(ValueError, match="Invalid log_target"): + parse_log_target("192.168.1.5") + + def test_non_digit_port_raises(self): + with pytest.raises(ValueError, match="Invalid log_target"): + parse_log_target("192.168.1.5:syslog") + + def test_empty_string_raises(self): + with pytest.raises(ValueError): + parse_log_target("") + + def test_multiple_colons_uses_last_as_port(self): + # IPv6-style or hostname with colons — rsplit takes the last segment + host, port = parse_log_target("::1:514") + assert port == 514 + + +class TestProbeLogTarget: + def test_returns_true_when_reachable(self): + mock_conn = MagicMock() + mock_conn.__enter__ = MagicMock(return_value=mock_conn) + mock_conn.__exit__ = MagicMock(return_value=False) + with patch("decnet.logging.forwarder.socket.create_connection", + return_value=mock_conn): + assert probe_log_target("192.168.1.5:5140") is True + + def test_returns_false_when_connection_refused(self): + with patch("decnet.logging.forwarder.socket.create_connection", + side_effect=OSError("Connection refused")): + assert probe_log_target("192.168.1.5:5140") is False + + def test_returns_false_on_timeout(self): + with patch("decnet.logging.forwarder.socket.create_connection", + side_effect=TimeoutError("timed out")): + assert probe_log_target("192.168.1.5:5140") is False + + def test_returns_false_on_invalid_target(self): + # ValueError from parse_log_target is caught and returns False + assert probe_log_target("not-a-valid-target") is False diff --git a/tests/test_mutator.py b/tests/test_mutator.py new file mode 100644 index 0000000..51afaf3 --- /dev/null +++ b/tests/test_mutator.py @@ -0,0 +1,207 @@ +""" +Tests for decnet.mutator — mutation engine, retry logic, due-time scheduling. +All subprocess and state I/O is mocked; no Docker or filesystem access. +""" +import subprocess +import time +from pathlib import Path +from unittest.mock import MagicMock, call, patch + +import pytest + +from decnet.config import DeckyConfig, DecnetConfig +from decnet.mutator import _compose_with_retry, mutate_all, mutate_decky + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_decky(name="decky-01", services=None, archetype=None, + mutate_interval=30, last_mutated=0.0): + return DeckyConfig( + name=name, + ip="192.168.1.10", + services=services or ["ssh"], + distro="debian", + base_image="debian", + hostname="host-01", + archetype=archetype, + mutate_interval=mutate_interval, + last_mutated=last_mutated, + ) + + +def _make_config(deckies=None, mutate_interval=30): + return DecnetConfig( + mode="unihost", interface="eth0", + subnet="192.168.1.0/24", gateway="192.168.1.1", + deckies=deckies or [_make_decky()], + mutate_interval=mutate_interval, + ) + + +# --------------------------------------------------------------------------- +# _compose_with_retry +# --------------------------------------------------------------------------- + +class TestComposeWithRetry: + def test_succeeds_on_first_attempt(self): + result = MagicMock(returncode=0, stdout="done\n") + with patch("decnet.mutator.subprocess.run", return_value=result) as mock_run: + _compose_with_retry("up", "-d", compose_file=Path("compose.yml")) + mock_run.assert_called_once() + + def test_retries_on_failure_then_succeeds(self): + fail = MagicMock(returncode=1, stdout="", stderr="transient error") + ok = MagicMock(returncode=0, stdout="", stderr="") + with patch("decnet.mutator.subprocess.run", side_effect=[fail, ok]) as mock_run, \ + patch("decnet.mutator.time.sleep"): + _compose_with_retry("up", "-d", compose_file=Path("compose.yml"), retries=3) + assert mock_run.call_count == 2 + + def test_raises_after_all_retries_exhausted(self): + fail = MagicMock(returncode=1, stdout="", stderr="hard error") + with patch("decnet.mutator.subprocess.run", return_value=fail), \ + patch("decnet.mutator.time.sleep"): + with pytest.raises(subprocess.CalledProcessError): + _compose_with_retry("up", "-d", compose_file=Path("compose.yml"), retries=3) + + def test_exponential_backoff(self): + fail = MagicMock(returncode=1, stdout="", stderr="") + sleep_calls = [] + with patch("decnet.mutator.subprocess.run", return_value=fail), \ + patch("decnet.mutator.time.sleep", side_effect=lambda d: sleep_calls.append(d)): + with pytest.raises(subprocess.CalledProcessError): + _compose_with_retry("up", compose_file=Path("c.yml"), retries=3, delay=1.0) + assert sleep_calls == [1.0, 2.0] + + def test_correct_command_structure(self): + ok = MagicMock(returncode=0, stdout="") + with patch("decnet.mutator.subprocess.run", return_value=ok) as mock_run: + _compose_with_retry("up", "-d", "--remove-orphans", + compose_file=Path("/tmp/compose.yml")) + cmd = mock_run.call_args[0][0] + assert cmd[:3] == ["docker", "compose", "-f"] + assert "up" in cmd + assert "--remove-orphans" in cmd + + +# --------------------------------------------------------------------------- +# mutate_decky +# --------------------------------------------------------------------------- + +class TestMutateDecky: + def _patch(self, config=None, compose_path=Path("compose.yml")): + """Return a context manager that mocks all I/O in mutate_decky.""" + cfg = config or _make_config() + return ( + patch("decnet.mutator.load_state", return_value=(cfg, compose_path)), + patch("decnet.mutator.save_state"), + patch("decnet.mutator.write_compose"), + patch("decnet.mutator._compose_with_retry"), + ) + + def test_returns_false_when_no_state(self): + with patch("decnet.mutator.load_state", return_value=None): + assert mutate_decky("decky-01") is False + + def test_returns_false_when_decky_not_found(self): + p = self._patch() + with p[0], p[1], p[2], p[3]: + assert mutate_decky("nonexistent") is False + + def test_returns_true_on_success(self): + p = self._patch() + with p[0], p[1], p[2], p[3]: + assert mutate_decky("decky-01") is True + + def test_saves_state_after_mutation(self): + p = self._patch() + with p[0], patch("decnet.mutator.save_state") as mock_save, p[2], p[3]: + mutate_decky("decky-01") + mock_save.assert_called_once() + + def test_regenerates_compose_after_mutation(self): + p = self._patch() + with p[0], p[1], patch("decnet.mutator.write_compose") as mock_compose, p[3]: + mutate_decky("decky-01") + mock_compose.assert_called_once() + + def test_returns_false_on_compose_failure(self): + p = self._patch() + err = subprocess.CalledProcessError(1, "docker", "", "compose failed") + with p[0], p[1], p[2], patch("decnet.mutator._compose_with_retry", side_effect=err): + assert mutate_decky("decky-01") is False + + def test_mutation_changes_services(self): + cfg = _make_config(deckies=[_make_decky(services=["ssh"])]) + p = self._patch(config=cfg) + with p[0], p[1], p[2], p[3]: + mutate_decky("decky-01") + # Services may have changed (or stayed the same after 20 attempts) + assert isinstance(cfg.deckies[0].services, list) + assert len(cfg.deckies[0].services) >= 1 + + def test_updates_last_mutated_timestamp(self): + cfg = _make_config(deckies=[_make_decky(last_mutated=0.0)]) + p = self._patch(config=cfg) + before = time.time() + with p[0], p[1], p[2], p[3]: + mutate_decky("decky-01") + assert cfg.deckies[0].last_mutated >= before + + def test_archetype_constrains_service_pool(self): + """A decky with an archetype must only mutate within its service pool.""" + cfg = _make_config(deckies=[_make_decky(archetype="workstation", services=["rdp"])]) + p = self._patch(config=cfg) + with p[0], p[1], p[2], p[3]: + result = mutate_decky("decky-01") + assert result is True + + +# --------------------------------------------------------------------------- +# mutate_all +# --------------------------------------------------------------------------- + +class TestMutateAll: + def test_no_state_returns_early(self): + with patch("decnet.mutator.load_state", return_value=None), \ + patch("decnet.mutator.mutate_decky") as mock_mutate: + mutate_all() + mock_mutate.assert_not_called() + + def test_force_mutates_all_deckies(self): + cfg = _make_config(deckies=[_make_decky("d1"), _make_decky("d2")]) + with patch("decnet.mutator.load_state", return_value=(cfg, Path("c.yml"))), \ + patch("decnet.mutator.mutate_decky", return_value=True) as mock_mutate: + mutate_all(force=True) + assert mock_mutate.call_count == 2 + + def test_skips_decky_not_yet_due(self): + # last_mutated = now, interval = 30 min → not due + now = time.time() + cfg = _make_config(deckies=[_make_decky(mutate_interval=30, last_mutated=now)]) + with patch("decnet.mutator.load_state", return_value=(cfg, Path("c.yml"))), \ + patch("decnet.mutator.mutate_decky") as mock_mutate: + mutate_all(force=False) + mock_mutate.assert_not_called() + + def test_mutates_decky_that_is_due(self): + # last_mutated = 2 hours ago, interval = 30 min → due + old_ts = time.time() - 7200 + cfg = _make_config(deckies=[_make_decky(mutate_interval=30, last_mutated=old_ts)]) + with patch("decnet.mutator.load_state", return_value=(cfg, Path("c.yml"))), \ + patch("decnet.mutator.mutate_decky", return_value=True) as mock_mutate: + mutate_all(force=False) + mock_mutate.assert_called_once_with("decky-01") + + def test_skips_decky_with_no_interval_and_no_force(self): + cfg = _make_config( + deckies=[_make_decky(mutate_interval=None)], + mutate_interval=None, + ) + with patch("decnet.mutator.load_state", return_value=(cfg, Path("c.yml"))), \ + patch("decnet.mutator.mutate_decky") as mock_mutate: + mutate_all(force=False) + mock_mutate.assert_not_called() diff --git a/tests/test_network.py b/tests/test_network.py index 193b1dd..af99534 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -10,12 +10,18 @@ from decnet.network import ( HOST_IPVLAN_IFACE, HOST_MACVLAN_IFACE, MACVLAN_NETWORK_NAME, + allocate_ips, create_ipvlan_network, create_macvlan_network, + detect_interface, + detect_subnet, + get_host_ip, ips_to_range, + remove_macvlan_network, setup_host_ipvlan, setup_host_macvlan, teardown_host_ipvlan, + teardown_host_macvlan, ) @@ -193,3 +199,154 @@ class TestSetupHostIpvlan: calls = [str(c) for c in mock_run.call_args_list] assert any(HOST_IPVLAN_IFACE in c for c in calls) assert not any(HOST_MACVLAN_IFACE in c for c in calls) + + +# --------------------------------------------------------------------------- +# allocate_ips (pure logic — no subprocess / Docker) +# --------------------------------------------------------------------------- + +class TestAllocateIps: + def test_basic_allocation(self): + ips = allocate_ips("192.168.1.0/24", "192.168.1.1", "192.168.1.100", count=3) + assert len(ips) == 3 + assert "192.168.1.1" not in ips # gateway skipped + assert "192.168.1.100" not in ips # host IP skipped + + def test_skips_network_and_broadcast(self): + ips = allocate_ips("10.0.0.0/30", "10.0.0.1", "10.0.0.3", count=1) + # /30 hosts: .1 (gateway), .2. .3 is host_ip → only .2 available + assert ips == ["10.0.0.2"] + + def test_respects_ip_start(self): + ips = allocate_ips("192.168.1.0/24", "192.168.1.1", "192.168.1.1", + count=2, ip_start="192.168.1.50") + assert all(ip >= "192.168.1.50" for ip in ips) + + def test_raises_when_not_enough_ips(self): + # /30 only has 2 host addresses; reserving both leaves 0 + with pytest.raises(RuntimeError, match="Not enough free IPs"): + allocate_ips("10.0.0.0/30", "10.0.0.1", "10.0.0.2", count=3) + + def test_no_duplicates(self): + ips = allocate_ips("10.0.0.0/24", "10.0.0.1", "10.0.0.2", count=10) + assert len(ips) == len(set(ips)) + + def test_exact_count_returned(self): + ips = allocate_ips("172.16.0.0/24", "172.16.0.1", "172.16.0.254", count=5) + assert len(ips) == 5 + + +# --------------------------------------------------------------------------- +# detect_interface +# --------------------------------------------------------------------------- + +class TestDetectInterface: + @patch("decnet.network._run") + def test_parses_dev_from_route(self, mock_run): + mock_run.return_value = MagicMock( + stdout="default via 192.168.1.1 dev eth0 proto dhcp\n" + ) + assert detect_interface() == "eth0" + + @patch("decnet.network._run") + def test_raises_when_no_dev_found(self, mock_run): + mock_run.return_value = MagicMock(stdout="") + with pytest.raises(RuntimeError, match="Could not auto-detect"): + detect_interface() + + +# --------------------------------------------------------------------------- +# detect_subnet +# --------------------------------------------------------------------------- + +class TestDetectSubnet: + def _make_run(self, addr_output, route_output): + def side_effect(cmd, **kwargs): + if "addr" in cmd: + return MagicMock(stdout=addr_output) + return MagicMock(stdout=route_output) + return side_effect + + @patch("decnet.network._run") + def test_parses_subnet_and_gateway(self, mock_run): + mock_run.side_effect = self._make_run( + " inet 192.168.1.5/24 brd 192.168.1.255 scope global eth0\n", + "default via 192.168.1.1 dev eth0\n", + ) + subnet, gw = detect_subnet("eth0") + assert subnet == "192.168.1.0/24" + assert gw == "192.168.1.1" + + @patch("decnet.network._run") + def test_raises_when_no_inet(self, mock_run): + mock_run.side_effect = self._make_run("", "default via 192.168.1.1 dev eth0\n") + with pytest.raises(RuntimeError, match="Could not detect subnet"): + detect_subnet("eth0") + + @patch("decnet.network._run") + def test_raises_when_no_gateway(self, mock_run): + mock_run.side_effect = self._make_run( + " inet 192.168.1.5/24 brd 192.168.1.255 scope global eth0\n", "" + ) + with pytest.raises(RuntimeError, match="Could not detect gateway"): + detect_subnet("eth0") + + +# --------------------------------------------------------------------------- +# get_host_ip +# --------------------------------------------------------------------------- + +class TestGetHostIp: + @patch("decnet.network._run") + def test_returns_host_ip(self, mock_run): + mock_run.return_value = MagicMock( + stdout=" inet 10.0.0.5/24 brd 10.0.0.255 scope global eth0\n" + ) + assert get_host_ip("eth0") == "10.0.0.5" + + @patch("decnet.network._run") + def test_raises_when_no_inet(self, mock_run): + mock_run.return_value = MagicMock(stdout="link/ether aa:bb:cc:dd:ee:ff\n") + with pytest.raises(RuntimeError, match="Could not determine host IP"): + get_host_ip("eth0") + + +# --------------------------------------------------------------------------- +# remove_macvlan_network +# --------------------------------------------------------------------------- + +class TestRemoveMacvlanNetwork: + def test_removes_matching_network(self): + client = MagicMock() + net = MagicMock() + net.name = MACVLAN_NETWORK_NAME + client.networks.list.return_value = [net] + remove_macvlan_network(client) + net.remove.assert_called_once() + + def test_noop_when_no_matching_network(self): + client = MagicMock() + other = MagicMock() + other.name = "some-other-network" + client.networks.list.return_value = [other] + remove_macvlan_network(client) + other.remove.assert_not_called() + + +# --------------------------------------------------------------------------- +# teardown_host_macvlan +# --------------------------------------------------------------------------- + +class TestTeardownHostMacvlan: + @patch("decnet.network.os.geteuid", return_value=0) + @patch("decnet.network._run") + def test_deletes_macvlan_iface(self, mock_run, _): + mock_run.return_value = MagicMock(returncode=0) + teardown_host_macvlan("192.168.1.96/27") + calls = [str(c) for c in mock_run.call_args_list] + assert any(HOST_MACVLAN_IFACE in c for c in calls) + + @patch("decnet.network.os.geteuid", return_value=1) + def test_requires_root(self, _): + with pytest.raises(PermissionError): + teardown_host_macvlan("192.168.1.96/27") diff --git a/tests/test_services.py b/tests/test_services.py index 0edb85b..4256da1 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -339,3 +339,25 @@ def test_redis_default_no_extra_env(): env = _fragment("redis").get("environment", {}) assert "REDIS_VERSION" not in env assert "REDIS_OS" not in env + + +# Telnet --------------------------------------------------------------------- + +def test_telnet_log_target_uses_cowrie_tcp_output(): + """Telnet forwards logs via Cowrie TCP output, same pattern as SSH.""" + env = _fragment("telnet", log_target="10.0.0.1:5140").get("environment", {}) + assert env.get("COWRIE_OUTPUT_TCP_ENABLED") == "true" + assert env.get("COWRIE_OUTPUT_TCP_HOST") == "10.0.0.1" + assert env.get("COWRIE_OUTPUT_TCP_PORT") == "5140" + + +def test_telnet_no_log_target_omits_tcp_output(): + env = _fragment("telnet").get("environment", {}) + assert "COWRIE_OUTPUT_TCP_ENABLED" not in env + assert "COWRIE_OUTPUT_TCP_HOST" not in env + + +def test_telnet_ssh_disabled_in_telnet_only_container(): + env = _fragment("telnet").get("environment", {}) + assert env.get("COWRIE_SSH_ENABLED") == "false" + assert env.get("COWRIE_TELNET_ENABLED") == "true"