87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
from types import SimpleNamespace
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from l4d2host.logs import stream_instance_logs
|
|
|
|
|
|
class DummyProcess:
|
|
def __init__(self, lines: list[str]) -> None:
|
|
self.stdout = SimpleNamespace(readline=self._readline)
|
|
self.stderr = SimpleNamespace(readline=lambda: "", read=lambda: "")
|
|
self._lines = iter(lines)
|
|
self.terminated = False
|
|
self.waited = False
|
|
|
|
def _readline(self) -> str:
|
|
return next(self._lines, "")
|
|
|
|
def poll(self):
|
|
return None if not self.waited else 0
|
|
|
|
def terminate(self) -> None:
|
|
self.terminated = True
|
|
|
|
def wait(self, timeout: int | None = None) -> None:
|
|
del timeout
|
|
self.waited = True
|
|
|
|
|
|
def test_stream_instance_logs_yields_lines(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
proc = DummyProcess(["line1\n", "line2\n", ""])
|
|
|
|
def fake_popen(cmd, **kwargs):
|
|
del cmd
|
|
del kwargs
|
|
return proc
|
|
|
|
monkeypatch.setattr("l4d2host.service_control.subprocess.Popen", fake_popen)
|
|
|
|
lines = list(stream_instance_logs("alpha", lines=10, follow=False))
|
|
assert lines == ["line1", "line2"]
|
|
assert proc.waited is True
|
|
|
|
|
|
def test_stream_instance_logs_uses_journalctl_helper(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_stream_command(cmd):
|
|
calls.append(list(cmd))
|
|
return iter(["line1"])
|
|
|
|
monkeypatch.setattr("l4d2host.service_control.stream_command", fake_stream_command)
|
|
|
|
assert list(stream_instance_logs("alpha", lines=25, follow=False)) == ["line1"]
|
|
assert calls == [
|
|
[
|
|
"sudo",
|
|
"-n",
|
|
"/usr/local/libexec/left4me/left4me-journalctl",
|
|
"alpha",
|
|
"--lines",
|
|
"25",
|
|
"--no-follow",
|
|
]
|
|
]
|
|
|
|
|
|
def test_stream_instance_logs_raises_when_helper_fails(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class FailedProcess:
|
|
stdout = SimpleNamespace(readline=lambda: "")
|
|
stderr = SimpleNamespace(read=lambda: "sudo denied\n")
|
|
|
|
def poll(self):
|
|
return 7
|
|
|
|
def wait(self, timeout: int | None = None):
|
|
del timeout
|
|
return 7
|
|
|
|
monkeypatch.setattr("l4d2host.service_control.subprocess.Popen", lambda cmd, **kwargs: FailedProcess())
|
|
|
|
with pytest.raises(subprocess.CalledProcessError) as excinfo:
|
|
list(stream_instance_logs("alpha", lines=10, follow=False))
|
|
|
|
assert excinfo.value.returncode == 7
|
|
assert excinfo.value.stderr == "sudo denied\n"
|