l4d2host/instances.py

@@ -8,7 +8,12 @@ from l4d2host.service_control import start_service, stop_service
 from l4d2host.spec import load_spec
 
 
-from l4d2host.logging import emit_step
+def _emit_step(msg: str, on_stdout: Callable[[str], None] | None, passthrough: bool) -> None:
+    formatted = f"Step: {msg}"
+    if on_stdout is not None:
+        on_stdout(formatted)
+    elif passthrough:
+        print(formatted, flush=True)
 
 
 DEFAULT_ROOT = DEFAULT_LEFT4ME_ROOT
@@ -27,7 +32,7 @@ def initialize_instance(
     root = get_left4me_root() if root is None else Path(root)
     spec = load_spec(spec_path)
 
-    emit_step("creating instance directories...", on_stdout, passthrough)
+    _emit_step("creating instance directories...", on_stdout, passthrough)
     instance_dir = root / "instances" / name
     runtime_dir = root / "runtime" / name
     (runtime_dir / "upper").mkdir(parents=True, exist_ok=True)
@@ -38,7 +43,7 @@ def initialize_instance(
     lowerdirs = [str(overlay_path(overlay, root=root)) for overlay in spec.overlays]
     lowerdirs.append(str(root / "installation"))
 
-    emit_step("writing instance.env...", on_stdout, passthrough)
+    _emit_step("writing instance.env...", on_stdout, passthrough)
     instance_env = "\n".join(
         [
             f"L4D2_PORT={spec.port}",
@@ -48,10 +53,10 @@ def initialize_instance(
     ) + "\n"
     (instance_dir / "instance.env").write_text(instance_env)
 
-    emit_step("writing server.cfg...", on_stdout, passthrough)
+    _emit_step("writing server.cfg...", on_stdout, passthrough)
     server_cfg = "\n".join(spec.config) if spec.config else ""
     (instance_dir / "server.cfg").write_text(server_cfg)
-    emit_step("initialization complete.", on_stdout, passthrough)
+    _emit_step("initialization complete.", on_stdout, passthrough)
 
 
 def _load_instance_env(path: Path) -> dict[str, str]:
@@ -79,7 +84,7 @@ def start_instance(
 
     env = _load_instance_env(instance_dir / "instance.env")
 
-    emit_step("mounting runtime overlay...", on_stdout, passthrough)
+    _emit_step("mounting runtime overlay...", on_stdout, passthrough)
     run_command(
         [
             "fuse-overlayfs",
@@ -97,12 +102,12 @@ def start_instance(
         should_cancel=should_cancel,
     )
 
-    emit_step("copying server.cfg to runtime...", on_stdout, passthrough)
+    _emit_step("copying server.cfg to runtime...", on_stdout, passthrough)
     target_cfg = runtime_dir / "merged" / "left4dead2" / "cfg" / "server.cfg"
     target_cfg.parent.mkdir(parents=True, exist_ok=True)
     shutil.copy2(instance_dir / "server.cfg", target_cfg)
 
-    emit_step("starting systemd service...", on_stdout, passthrough)
+    _emit_step("starting systemd service...", on_stdout, passthrough)
     start_service(
         name,
         on_stdout=on_stdout,
@@ -110,7 +115,7 @@ def start_instance(
         passthrough=passthrough,
         should_cancel=should_cancel,
     )
-    emit_step("start complete.", on_stdout, passthrough)
+    _emit_step("start complete.", on_stdout, passthrough)
 
 
 def stop_instance(
@@ -123,7 +128,7 @@ def stop_instance(
     should_cancel: Callable[[], bool] | None = None,
 ) -> None:
     root = get_left4me_root() if root is None else Path(root)
-    emit_step("stopping systemd service...", on_stdout, passthrough)
+    _emit_step("stopping systemd service...", on_stdout, passthrough)
     stop_service(
         name,
         on_stdout=on_stdout,
@@ -131,7 +136,7 @@ def stop_instance(
         passthrough=passthrough,
         should_cancel=should_cancel,
     )
-    emit_step("unmounting runtime overlay...", on_stdout, passthrough)
+    _emit_step("unmounting runtime overlay...", on_stdout, passthrough)
     run_command(
         ["fusermount3", "-u", str(root / "runtime" / name / "merged")],
         on_stdout=on_stdout,
@@ -139,7 +144,7 @@ def stop_instance(
         passthrough=passthrough,
         should_cancel=should_cancel,
     )
-    emit_step("stop complete.", on_stdout, passthrough)
+    _emit_step("stop complete.", on_stdout, passthrough)
 
 
 def delete_instance(
@@ -158,7 +163,7 @@ def delete_instance(
     if not instance_dir.exists() and not runtime_dir.exists():
         return
 
-    emit_step("stopping systemd service (if running)...", on_stdout, passthrough)
+    _emit_step("stopping systemd service (if running)...", on_stdout, passthrough)
     stop_service(
         name,
         on_stdout=on_stdout,
@@ -169,7 +174,7 @@ def delete_instance(
 
     merged = runtime_dir / "merged"
     if merged.is_mount():
-        emit_step("unmounting runtime overlay (if mounted)...", on_stdout, passthrough)
+        _emit_step("unmounting runtime overlay (if mounted)...", on_stdout, passthrough)
         run_command(
             ["fusermount3", "-u", str(merged)],
             on_stdout=on_stdout,
@@ -178,9 +183,9 @@ def delete_instance(
             should_cancel=should_cancel,
         )
 
-    emit_step("removing instance files...", on_stdout, passthrough)
+    _emit_step("removing instance files...", on_stdout, passthrough)
     if instance_dir.exists():
         shutil.rmtree(instance_dir)
     if runtime_dir.exists():
         shutil.rmtree(runtime_dir)
-    emit_step("delete complete.", on_stdout, passthrough)
+    _emit_step("delete complete.", on_stdout, passthrough)

l4d2host/logging.py

@@ -1,8 +0,0 @@
-from typing import Callable
-
-def emit_step(msg: str, on_stdout: Callable[[str], None] | None, passthrough: bool) -> None:
-    formatted = f"Step: {msg}"
-    if on_stdout is not None:
-        on_stdout(formatted)
-    if passthrough:
-        print(formatted, flush=True)

l4d2host/process.py

@@ -46,7 +46,7 @@ def run_command(
         if on_stderr is not None:
             on_stderr(line)
         if passthrough:
-            print(line, file=sys.stderr, flush=True)
+            print(line, file=sys.stderr)
 
     def terminate_process() -> None:
         emit_stderr_message("cancellation requested; terminating subprocess")
@@ -82,7 +82,7 @@ def run_command(
             if callback is not None:
                 callback(line)
             if passthrough:
-                print(line, file=output_stream, flush=True)
+                print(line, file=output_stream)
         stream.close()
 
     stdout_thread = threading.Thread(

l4d2host/steam_install.py

@@ -3,7 +3,6 @@ from typing import Callable
 
 from l4d2host.paths import get_left4me_root
 from l4d2host.process import run_command
-from l4d2host.logging import emit_step
 
 
 class SteamInstaller:
@@ -20,7 +19,6 @@ class SteamInstaller:
         should_cancel: Callable[[], bool] | None = None,
     ) -> None:
         for platform in ("windows", "linux"):
-            emit_step(f"downloading {platform} platform payload...", on_stdout, passthrough)
             run_command(
                 [
                     self.steamcmd,
@@ -40,4 +38,3 @@ class SteamInstaller:
                 passthrough=passthrough,
                 should_cancel=should_cancel,
             )
-        emit_step("installation complete.", on_stdout, passthrough)

l4d2host/tests/test_initialize.py

@@ -2,7 +2,7 @@ import sys
 from io import StringIO
 from pathlib import Path
 
-from l4d2host.instances import initialize_instance
+from l4d2host.instances import initialize_instance, _emit_step
 
 
 def test_initialize_writes_files(tmp_path: Path) -> None:
@@ -35,6 +35,23 @@ def test_initialize_uses_configured_left4me_root(tmp_path: Path, monkeypatch) ->
     assert f"L4D2_LOWERDIRS={tmp_path}/overlays/standard:{tmp_path}/installation" in env
 
 
+def test_emit_step_uses_callback() -> None:
+    calls: list[str] = []
+    _emit_step("test step", on_stdout=calls.append, passthrough=False)
+    assert calls == ["Step: test step"]
+
+
+def test_emit_step_uses_passthrough_stdout(monkeypatch) -> None:
+    fake_out = StringIO()
+    monkeypatch.setattr(sys, "stdout", fake_out)
+    _emit_step("passthrough step", on_stdout=None, passthrough=True)
+    assert fake_out.getvalue() == "Step: passthrough step\n"
+
+
+def test_emit_step_does_nothing_if_no_target() -> None:
+    _emit_step("silent step", on_stdout=None, passthrough=False)
+
+
 def test_initialize_instance_emits_steps(tmp_path: Path) -> None:
     spec = tmp_path / "spec.yaml"
     spec.write_text("port: 27015\noverlays: [standard]\n")

l4d2host/tests/test_install.py

@@ -43,20 +43,3 @@ def test_default_install_dir_uses_left4me_root(tmp_path, monkeypatch):
     SteamInstaller().install_or_update()
 
     assert str(tmp_path / "installation") in calls[0]
-
-
-def test_steam_installer_emits_steps(tmp_path, monkeypatch) -> None:
-    from pathlib import Path
-    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
-    monkeypatch.setattr("l4d2host.steam_install.run_command", lambda cmd, **kwargs: None)
-
-    steps: list[str] = []
-
-    from l4d2host.steam_install import SteamInstaller
-    SteamInstaller().install_or_update(on_stdout=steps.append)
-
-    assert steps == [
-        "Step: downloading windows platform payload...",
-        "Step: downloading linux platform payload...",
-        "Step: installation complete."
-    ]

l4d2host/tests/test_logging.py

@@ -1,30 +0,0 @@
-import sys
-from io import StringIO
-
-from l4d2host.logging import emit_step
-
-
-def test_emit_step_uses_callback() -> None:
-    calls: list[str] = []
-    emit_step("test step", on_stdout=calls.append, passthrough=False)
-    assert calls == ["Step: test step"]
-
-
-def test_emit_step_uses_passthrough_stdout(monkeypatch) -> None:
-    fake_out = StringIO()
-    monkeypatch.setattr(sys, "stdout", fake_out)
-    emit_step("passthrough step", on_stdout=None, passthrough=True)
-    assert fake_out.getvalue() == "Step: passthrough step\n"
-
-
-def test_emit_step_does_nothing_if_no_target() -> None:
-    emit_step("silent step", on_stdout=None, passthrough=False)
-
-
-def test_emit_step_does_both(monkeypatch) -> None:
-    calls: list[str] = []
-    fake_out = StringIO()
-    monkeypatch.setattr(sys, "stdout", fake_out)
-    emit_step("both step", on_stdout=calls.append, passthrough=True)
-    assert calls == ["Step: both step"]
-    assert fake_out.getvalue() == "Step: both step\n"

l4d2host/tests/test_process.py

@@ -47,21 +47,3 @@ def test_cancelled_command_raises_cancelled_error() -> None:
 def test_run_command_avoids_runtime_unsafe_nested_annotations() -> None:
     source = inspect.getsource(run_command)
     assert "subprocess.Popen[str].stdout" not in source
-
-
-def test_run_command_passthrough_writes_to_sys_streams(monkeypatch: pytest.MonkeyPatch) -> None:
-    import sys
-    from io import StringIO
-
-    mock_stdout = StringIO()
-    mock_stderr = StringIO()
-    monkeypatch.setattr(sys, "stdout", mock_stdout)
-    monkeypatch.setattr(sys, "stderr", mock_stderr)
-
-    run_command(
-        ["python3", "-c", "import sys; print('passed out'); print('passed err', file=sys.stderr)"],
-        passthrough=True,
-    )
-
-    assert mock_stdout.getvalue() == "passed out\n"
-    assert mock_stderr.getvalue() == "passed err\n"

l4d2web/services/host_commands.py

@@ -50,7 +50,7 @@ def run_command(
         if on_stderr is not None:
             on_stderr(line)
         if passthrough:
-            print(line, file=sys.stderr, flush=True)
+            print(line, file=sys.stderr)
 
     def terminate_process() -> None:
         emit_stderr_message("cancellation requested; terminating subprocess")
@@ -86,7 +86,7 @@ def run_command(
             if callback is not None:
                 callback(line)
             if passthrough:
-                print(line, file=output_stream, flush=True)
+                print(line, file=output_stream)
         stream.close()
 
     stdout_thread = threading.Thread(

l4d2web/tests/test_host_commands.py

@@ -1,25 +1,6 @@
-import sys
-from io import StringIO
 import pytest
 
 
-def test_run_command_passthrough_writes_to_sys_streams(monkeypatch: pytest.MonkeyPatch) -> None:
-    from l4d2web.services.host_commands import run_command
-
-    mock_stdout = StringIO()
-    mock_stderr = StringIO()
-    monkeypatch.setattr(sys, "stdout", mock_stdout)
-    monkeypatch.setattr(sys, "stderr", mock_stderr)
-
-    run_command(
-        ["python3", "-c", "import sys; print('passed out'); print('passed err', file=sys.stderr)"],
-        passthrough=True,
-    )
-
-    assert mock_stdout.getvalue() == "passed out\n"
-    assert mock_stderr.getvalue() == "passed err\n"
-
-
 def test_run_command_streams_stdout_and_stderr_callbacks() -> None:
     from l4d2web.services.host_commands import run_command