530 lines
22 KiB
Python
530 lines
22 KiB
Python
"""Host network settings (spec §15): agent ops, gating, web API.
|
|
|
|
Two layers:
|
|
* the agent protocol itself — validation, whitelist refusals,
|
|
auto-snapshot + restore, audit trail, config loading (FakeBackend;
|
|
the Linux backend shares the same dispatch in `handle`)
|
|
* the web API — the FR-15.2.1/2 gates (404 when off), readonly 403,
|
|
mutation mapping through an InProcessClient-backed HostNetService
|
|
(the socket path is exercised only on real Linux hosts).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from wago_sim.hostnet import (FakeBackend, HostNetAgent, HostNetError,
|
|
HostNetRefused, HostNetService,
|
|
HostNetUnavailable, InProcessClient,
|
|
hostnet_enabled, load_agent_config,
|
|
parse_iface, parse_ipv4, parse_prefix)
|
|
from wago_sim.webui.app import build_webui
|
|
|
|
|
|
@pytest.fixture
|
|
def agent():
|
|
"""Agent with two whitelisted interfaces on the fake backend."""
|
|
return HostNetAgent(
|
|
backend=FakeBackend({"eth0": ["192.168.1.10/24"],
|
|
"eth1": ["10.0.0.5/8"],
|
|
"lo": ["127.0.0.1/8"]}),
|
|
whitelist=["eth0", "eth1"],
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def service(agent):
|
|
return HostNetService(InProcessClient(agent))
|
|
|
|
|
|
@pytest.fixture
|
|
async def hostnet_client(service):
|
|
"""httpx client against a web UI with the hostnet service mounted."""
|
|
from wago_sim.app import build_app
|
|
from wago_sim.modules import DI_8, DO_8
|
|
|
|
sim = build_app(modules=[DI_8, DO_8])
|
|
webui = build_webui(sim, user="admin", password="simu",
|
|
hostnet_service=service)
|
|
import httpx
|
|
|
|
transport = httpx.ASGITransport(app=webui)
|
|
async with httpx.AsyncClient(transport=transport,
|
|
base_url="http://test") as c:
|
|
r = await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
assert r.status_code == 200
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def plain_client():
|
|
"""httpx client against a web UI withOUT the hostnet service."""
|
|
from wago_sim.app import build_app
|
|
from wago_sim.modules import DI_8, DO_8
|
|
|
|
sim = build_app(modules=[DI_8, DO_8])
|
|
webui = build_webui(sim, user="admin", password="simu")
|
|
import httpx
|
|
|
|
transport = httpx.ASGITransport(app=webui)
|
|
async with httpx.AsyncClient(transport=transport,
|
|
base_url="http://test") as c:
|
|
r = await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
assert r.status_code == 200
|
|
yield c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation helpers (FR-15.2.4 — used by both the API and the agent)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValidation:
|
|
def test_ipv4_normalized(self):
|
|
assert parse_ipv4(" 192.168.1.33 ") == "192.168.1.33"
|
|
|
|
@pytest.mark.parametrize("bad", ["banana", "1.2.3", "1.2.3.999", "",
|
|
None, 33, "192.168.1.1/24"])
|
|
def test_ipv4_rejected(self, bad):
|
|
with pytest.raises(HostNetError):
|
|
parse_ipv4(bad)
|
|
|
|
def test_prefix_bounds(self):
|
|
assert parse_prefix(24) == 24
|
|
assert parse_prefix("28") == 28
|
|
for bad in (0, 33, "x", None, -1):
|
|
with pytest.raises(HostNetError):
|
|
parse_prefix(bad)
|
|
|
|
def test_iface_name_shape(self):
|
|
assert parse_iface("eth0") == "eth0"
|
|
assert parse_iface("enp3s0") == "enp3s0"
|
|
assert parse_iface("br-123_4") == "br-123_4"
|
|
for bad in ("", None, "eth 0", "eth0; rm -rf /", "x" * 16, "/proc/x"):
|
|
with pytest.raises(HostNetError):
|
|
parse_iface(bad)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent protocol (FR-15.2.4 / 15.2.5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAgentOps:
|
|
def test_health_reports_whitelist(self, agent):
|
|
r = agent.handle({"op": "health"})
|
|
assert r["ok"] is True
|
|
assert r["whitelist"] == ["eth0", "eth1"]
|
|
assert r["protocol"] == 1
|
|
|
|
def test_list_is_live_state(self, agent):
|
|
r = agent.handle({"op": "list"})
|
|
names = {i["name"] for i in r["interfaces"]}
|
|
assert names == {"eth0", "eth1", "lo"}
|
|
eth0 = next(i for i in r["interfaces"] if i["name"] == "eth0")
|
|
assert eth0["addresses"] == [{"address": "192.168.1.10", "prefix": 24}]
|
|
|
|
def test_add_address(self, agent):
|
|
r = agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "192.168.1.33", "prefix": 24})
|
|
assert r["ok"] is True
|
|
eth0 = next(i for i in agent.handle({"op": "list"})["interfaces"]
|
|
if i["name"] == "eth0")
|
|
assert {"address": "192.168.1.33", "prefix": 24} in eth0["addresses"]
|
|
|
|
def test_add_duplicate_rejected(self, agent):
|
|
agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "192.168.1.33", "prefix": 24})
|
|
r = agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "192.168.1.33", "prefix": 24})
|
|
assert r["ok"] is False
|
|
assert "already exists" in r["error"]
|
|
|
|
def test_remove_address(self, agent):
|
|
agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "192.168.1.33", "prefix": 24})
|
|
r = agent.handle({"op": "remove_address", "iface": "eth0",
|
|
"address": "192.168.1.33"})
|
|
assert r["ok"] is True
|
|
eth0 = next(i for i in agent.handle({"op": "list"})["interfaces"]
|
|
if i["name"] == "eth0")
|
|
assert all(a["address"] != "192.168.1.33" for a in eth0["addresses"])
|
|
|
|
def test_remove_missing_rejected(self, agent):
|
|
r = agent.handle({"op": "remove_address", "iface": "eth0",
|
|
"address": "9.9.9.9"})
|
|
assert r["ok"] is False
|
|
assert "not assigned" in r["error"]
|
|
|
|
def test_change_address(self, agent):
|
|
r = agent.handle({"op": "change_address", "iface": "eth0",
|
|
"old": "192.168.1.10", "new": "192.168.1.20",
|
|
"prefix": 24})
|
|
assert r["ok"] is True
|
|
eth0 = next(i for i in agent.handle({"op": "list"})["interfaces"]
|
|
if i["name"] == "eth0")
|
|
assert {"address": "192.168.1.20", "prefix": 24} in eth0["addresses"]
|
|
assert all(a["address"] != "192.168.1.10" for a in eth0["addresses"])
|
|
|
|
def test_whitelist_refusal(self, agent):
|
|
# lo exists but is NOT whitelisted -> refused + audited
|
|
r = agent.handle({"op": "add_address", "iface": "lo",
|
|
"address": "127.0.0.2", "prefix": 8})
|
|
assert r["ok"] is False
|
|
assert r.get("refused") is True
|
|
assert "whitelist" in r["error"]
|
|
|
|
def test_unknown_interface_refused(self, agent):
|
|
r = agent.handle({"op": "snapshot", "iface": "wlan9"})
|
|
assert r["ok"] is False
|
|
assert r.get("refused") is True
|
|
|
|
def test_unknown_op(self, agent):
|
|
r = agent.handle({"op": "reboot_host"})
|
|
assert r["ok"] is False
|
|
assert "unknown operation" in r["error"]
|
|
|
|
def test_invalid_input_rejected(self, agent):
|
|
r = agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "not-an-ip", "prefix": 24})
|
|
assert r["ok"] is False
|
|
assert "invalid IPv4" in r["error"]
|
|
|
|
|
|
class TestSnapshotRestore:
|
|
"""FR-15.2.5: snapshot before every mutation; restore walks back."""
|
|
|
|
def _addresses(self, agent, iface):
|
|
return [a["address"] for i in agent.handle({"op": "list"})["interfaces"]
|
|
if i["name"] == iface for a in i["addresses"]]
|
|
|
|
def test_auto_snapshot_on_add(self, agent):
|
|
before = self._addresses(agent, "eth0")
|
|
agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "192.168.1.33", "prefix": 24})
|
|
# restore walks back to the pre-add state
|
|
r = agent.handle({"op": "restore", "iface": "eth0"})
|
|
assert r["ok"] is True
|
|
assert self._addresses(agent, "eth0") == before
|
|
|
|
def test_restore_after_remove(self, agent):
|
|
agent.handle({"op": "remove_address", "iface": "eth0",
|
|
"address": "192.168.1.10"})
|
|
assert "192.168.1.10" not in self._addresses(agent, "eth0")
|
|
agent.handle({"op": "restore", "iface": "eth0"})
|
|
assert "192.168.1.10" in self._addresses(agent, "eth0")
|
|
|
|
def test_restore_without_snapshot_rejected(self, agent):
|
|
r = agent.handle({"op": "restore", "iface": "eth1"})
|
|
# eth1 had no mutation yet -> nothing to restore
|
|
assert r["ok"] is False
|
|
assert "no snapshot" in r["error"]
|
|
|
|
def test_explicit_snapshot_op(self, agent):
|
|
r = agent.handle({"op": "snapshot", "iface": "eth0"})
|
|
assert r["ok"] is True
|
|
assert [a[0] for a in r["addresses"]] == ["192.168.1.10"]
|
|
|
|
|
|
class TestAudit:
|
|
def test_mutations_and_refusals_audited(self, agent):
|
|
agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "192.168.1.33", "prefix": 24,
|
|
"operator": "admin"})
|
|
agent.handle({"op": "add_address", "iface": "lo",
|
|
"address": "127.0.0.2", "prefix": 8})
|
|
lines = agent.handle({"op": "audit", "limit": 50})["lines"]
|
|
joined = "\n".join(lines)
|
|
assert "add eth0 192.168.1.33/24 by admin" in joined
|
|
assert "REFUSED" in joined
|
|
assert "whitelist" in joined
|
|
|
|
def test_audit_bounded(self, agent):
|
|
for i in range(10):
|
|
agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": f"10.60.{i}.1", "prefix": 24})
|
|
lines = agent.handle({"op": "audit", "limit": 5})["lines"]
|
|
assert len(lines) == 5
|
|
|
|
def test_operator_tagged(self, agent):
|
|
agent.handle({"op": "add_address", "iface": "eth0",
|
|
"address": "10.9.9.1", "prefix": 24,
|
|
"operator": "commissioner"})
|
|
joined = "\n".join(agent.handle({"op": "audit"})["lines"])
|
|
assert "by commissioner" in joined
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gating (FR-15.2.1 / 15.2.2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGating:
|
|
def test_off_without_flag(self, tmp_path):
|
|
assert hostnet_enabled({"WAGO_SIM_HOSTNET": ""}, []) is False
|
|
assert hostnet_enabled({}, []) is False
|
|
|
|
def test_flag_alone_is_not_enough(self, tmp_path):
|
|
# flag set, docker markers absent, no env marker -> off
|
|
assert hostnet_enabled(
|
|
{"WAGO_SIM_HOSTNET": "1", "WAGO_SIM_IN_DOCKER": "0"}, []) is False
|
|
|
|
def test_flag_and_docker_env(self):
|
|
assert hostnet_enabled(
|
|
{"WAGO_SIM_HOSTNET": "1", "WAGO_SIM_IN_DOCKER": "1"}, []) is True
|
|
|
|
def test_flag_and_marker_file(self, tmp_path):
|
|
marker = tmp_path / ".dockerenv"
|
|
marker.write_text("")
|
|
assert hostnet_enabled({"WAGO_SIM_HOSTNET": "true"}, [marker]) is True
|
|
|
|
@pytest.mark.parametrize("value", ["1", "true", "yes", "on", "YES", " On "])
|
|
def test_flag_truthy_forms(self, value):
|
|
assert hostnet_enabled(
|
|
{"WAGO_SIM_HOSTNET": value, "WAGO_SIM_IN_DOCKER": "1"}, []) is True
|
|
|
|
def test_flag_falsy_form(self):
|
|
assert hostnet_enabled(
|
|
{"WAGO_SIM_HOSTNET": "0", "WAGO_SIM_IN_DOCKER": "1"}, []) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent config + startup rules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAgentConfig:
|
|
def test_valid_config(self, tmp_path):
|
|
cfg_file = tmp_path / "agent.yaml"
|
|
cfg_file.write_text(
|
|
"socket: /tmp/x.sock\nwhitelist: [eth0, eth1]\naudit_path: /tmp/a.log\n")
|
|
cfg = load_agent_config(str(cfg_file))
|
|
assert cfg["socket"] == "/tmp/x.sock"
|
|
assert cfg["whitelist"] == ["eth0", "eth1"]
|
|
|
|
def test_missing_file(self):
|
|
with pytest.raises(HostNetError):
|
|
load_agent_config("nope/missing.yaml")
|
|
|
|
def test_empty_whitelist_refused_at_startup(self, tmp_path):
|
|
# a config without a whitelist must not start the agent
|
|
cfg_file = tmp_path / "agent.yaml"
|
|
cfg_file.write_text("socket: /tmp/x.sock\nwhitelist: []\n")
|
|
with pytest.raises(HostNetError):
|
|
HostNetAgent(backend=FakeBackend(), whitelist=[])
|
|
|
|
def test_bad_yaml(self, tmp_path):
|
|
cfg_file = tmp_path / "agent.yaml"
|
|
cfg_file.write_text("socket: [unclosed\n")
|
|
with pytest.raises(HostNetError):
|
|
load_agent_config(str(cfg_file))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Web API (spec §15.3 surface)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHostnetAPI:
|
|
async def test_disabled_routes_are_404(self, plain_client):
|
|
# FR-15.2.1/2: no page, no API surface when the gates are off
|
|
r = await plain_client.get("/api/hostnet")
|
|
assert r.status_code == 404
|
|
r = await plain_client.get("/api/hostnet/audit")
|
|
assert r.status_code == 404
|
|
r = await plain_client.post("/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "10.1.1.1", "prefix": 24})
|
|
assert r.status_code == 404
|
|
|
|
async def test_status_and_list(self, hostnet_client):
|
|
r = await hostnet_client.get("/api/hostnet")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["status"]["available"] is True
|
|
assert data["status"]["agent"]["whitelist"] == ["eth0", "eth1"]
|
|
names = {i["name"] for i in data["interfaces"]}
|
|
assert names == {"eth0", "eth1", "lo"}
|
|
|
|
async def test_add_then_remove_via_api(self, hostnet_client):
|
|
r = await hostnet_client.post(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "192.168.1.33", "prefix": 24})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["address"] == "192.168.1.33"
|
|
r = await hostnet_client.delete(
|
|
"/api/hostnet/interfaces/eth0/addresses/192.168.1.33")
|
|
assert r.status_code == 200
|
|
|
|
async def test_change_via_api(self, hostnet_client):
|
|
r = await hostnet_client.put(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"old": "192.168.1.10", "new": "192.168.1.20",
|
|
"prefix": 24})
|
|
assert r.status_code == 200
|
|
r = await hostnet_client.get("/api/hostnet")
|
|
eth0 = next(i for i in r.json()["interfaces"] if i["name"] == "eth0")
|
|
assert {"address": "192.168.1.20", "prefix": 24} in eth0["addresses"]
|
|
|
|
async def test_restore_via_api(self, hostnet_client):
|
|
await hostnet_client.post(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "192.168.1.33", "prefix": 24})
|
|
r = await hostnet_client.post("/api/hostnet/interfaces/eth0/restore")
|
|
assert r.status_code == 200
|
|
r = await hostnet_client.get("/api/hostnet")
|
|
eth0 = next(i for i in r.json()["interfaces"] if i["name"] == "eth0")
|
|
assert all(a["address"] != "192.168.1.33" for a in eth0["addresses"])
|
|
|
|
async def test_bad_input_400(self, hostnet_client):
|
|
r = await hostnet_client.post(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "banana", "prefix": 24})
|
|
assert r.status_code == 400
|
|
assert "invalid IPv4" in r.json()["detail"]
|
|
|
|
async def test_non_whitelisted_403(self, hostnet_client):
|
|
r = await hostnet_client.post(
|
|
"/api/hostnet/interfaces/lo/addresses",
|
|
json={"address": "127.0.0.2", "prefix": 8})
|
|
assert r.status_code == 403
|
|
assert "whitelist" in r.json()["detail"]
|
|
|
|
async def test_audit_view(self, hostnet_client):
|
|
await hostnet_client.post(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "192.168.1.33", "prefix": 24})
|
|
r = await hostnet_client.get("/api/hostnet/audit")
|
|
assert r.status_code == 200
|
|
assert "192.168.1.33" in "\n".join(r.json()["lines"])
|
|
|
|
async def test_operator_in_audit_trail(self, hostnet_client):
|
|
await hostnet_client.post(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "10.77.0.1", "prefix": 24})
|
|
r = await hostnet_client.get("/api/hostnet/audit")
|
|
# the logged-in user (admin) is named in every mutation line
|
|
assert "by admin" in "\n".join(r.json()["lines"])
|
|
|
|
|
|
class TestHostnetReadonly:
|
|
@pytest.fixture
|
|
async def ro_client(self, agent):
|
|
from wago_sim.app import build_app
|
|
from wago_sim.modules import DI_8, DO_8
|
|
import httpx
|
|
|
|
sim = build_app(modules=[DI_8, DO_8])
|
|
service = HostNetService(InProcessClient(agent))
|
|
webui = build_webui(sim, user="admin", password="simu",
|
|
readonly=True, hostnet_service=service)
|
|
transport = httpx.ASGITransport(app=webui)
|
|
async with httpx.AsyncClient(transport=transport,
|
|
base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
yield c
|
|
|
|
async def test_read_status_allowed(self, ro_client):
|
|
r = await ro_client.get("/api/hostnet")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"]["available"] is True
|
|
|
|
async def test_mutations_403_in_readonly(self, ro_client):
|
|
r = await ro_client.post(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "192.168.1.33", "prefix": 24})
|
|
assert r.status_code == 403
|
|
r = await ro_client.delete(
|
|
"/api/hostnet/interfaces/eth0/addresses/192.168.1.10")
|
|
assert r.status_code == 403
|
|
r = await ro_client.put(
|
|
"/api/hostnet/interfaces/eth0/addresses",
|
|
json={"old": "192.168.1.10", "new": "1.2.3.4", "prefix": 24})
|
|
assert r.status_code == 403
|
|
r = await ro_client.post("/api/hostnet/interfaces/eth0/restore")
|
|
assert r.status_code == 403
|
|
|
|
|
|
class TestUnavailableAgent:
|
|
async def test_unavailable_reported_not_raised(self):
|
|
"""FR-15.2.8: agent down -> available:false, no 5xx, banner text."""
|
|
from wago_sim.app import build_app
|
|
from wago_sim.modules import DI_8, DO_8
|
|
import httpx
|
|
|
|
class DeadClient:
|
|
"""Agent down: every op raises HostNetUnavailable (FR-15.2.8)."""
|
|
def health(self):
|
|
raise HostNetUnavailable("socket not found")
|
|
def list(self):
|
|
raise HostNetUnavailable("socket not found")
|
|
def add_address(self, iface, address, prefix, operator=""):
|
|
raise HostNetUnavailable("socket not found")
|
|
def remove_address(self, iface, address, operator=""):
|
|
raise HostNetUnavailable("socket not found")
|
|
def change_address(self, iface, old, new, prefix, operator=""):
|
|
raise HostNetUnavailable("socket not found")
|
|
def restore(self, iface, operator=""):
|
|
raise HostNetUnavailable("socket not found")
|
|
def audit(self, limit=50):
|
|
raise HostNetUnavailable("socket not found")
|
|
|
|
sim = build_app(modules=[DI_8, DO_8])
|
|
webui = build_webui(sim, user="admin", password="simu",
|
|
hostnet_service=HostNetService(DeadClient()))
|
|
transport = httpx.ASGITransport(app=webui)
|
|
async with httpx.AsyncClient(transport=transport,
|
|
base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
r = await c.get("/api/hostnet")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["status"]["available"] is False
|
|
assert "socket not found" in data["status"]["error"]
|
|
assert data["interfaces"] is None
|
|
# mutations report 503 with the same message
|
|
r = await c.post("/api/hostnet/interfaces/eth0/addresses",
|
|
json={"address": "1.2.3.4", "prefix": 24})
|
|
assert r.status_code == 503
|
|
|
|
async def test_hostnet_client_without_socket(self, tmp_path):
|
|
from wago_sim.hostnet import HostNetClient
|
|
|
|
client = HostNetClient(socket_path=str(tmp_path / "missing.sock"))
|
|
with pytest.raises(HostNetUnavailable):
|
|
client.health()
|
|
|
|
|
|
class TestIproute2ParseFallback:
|
|
"""The `ip -o` output parsers — the fallback read path."""
|
|
|
|
def test_addr_parse(self, monkeypatch):
|
|
monkeypatch.setenv("PATH", "/nonexistent") # `ip` never found
|
|
from wago_sim.hostnet import Iproute2Backend
|
|
|
|
backend = Iproute2Backend.__new__(Iproute2Backend)
|
|
backend._ipr = None
|
|
sample = ("2: eth0 inet 192.168.1.5/24 brd 192.168.1.255 "
|
|
"scope global eth0\n"
|
|
"3: eth1 inet 10.0.0.9/8 brd 10.255.255.255 "
|
|
"scope global eth1\n"
|
|
"4: docker0 inet 172.17.0.1/16 brd 172.17.255.255 "
|
|
"scope global docker0\n")
|
|
monkeypatch.setattr(backend, "_ip_cmd", lambda cmd: sample)
|
|
addrs = backend._addr_list()
|
|
assert ("eth0", "192.168.1.5", 24) in addrs
|
|
assert ("eth1", "10.0.0.9", 8) in addrs
|
|
|
|
def test_link_parse(self, monkeypatch):
|
|
monkeypatch.setenv("PATH", "/nonexistent")
|
|
from wago_sim.hostnet import Iproute2Backend
|
|
|
|
backend = Iproute2Backend.__new__(Iproute2Backend)
|
|
backend._ipr = None
|
|
sample = ("1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue\n"
|
|
"2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
|
|
"3: eth9: <BROADCAST,MULTICAST> mtu 1500\n")
|
|
monkeypatch.setattr(backend, "_ip_cmd", lambda cmd: sample)
|
|
links = backend._link_list()
|
|
assert ("lo", True) in links
|
|
assert ("eth0", True) in links
|
|
assert ("eth9", False) in links |