42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
from types import SimpleNamespace
|
|
|
|
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: "")
|
|
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:
|
|
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.logs.subprocess.Popen", fake_popen)
|
|
|
|
lines = list(stream_instance_logs("alpha", lines=10, follow=False))
|
|
assert lines == ["line1", "line2"]
|
|
assert proc.terminated is True
|