anti/DECNET
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

test: expand coverage 64%→76%; add BUGS.md for Gemini migration issues

Browse Source
This commit is contained in:
anti
2026-04-09 12:55:52 -04:00
parent 310c2a1fbe
commit 4ae6f4f23d
7 changed files with 717 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
tests/test_config.py Normal file
View File

@@ -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

65
tests/test_custom_service.py Normal file
View File

@@ -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

62
tests/test_logging_forwarder.py Normal file
View File

@@ -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

207
tests/test_mutator.py Normal file
View File

@@ -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()

157
tests/test_network.py
View File

@@ -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")

22
tests/test_services.py
View File

@@ -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"