feat: wire muxterm into FastAPI lifespan (start on boot, stop on shutdown)

muxplex/tests/test_lifespan_muxterm.py

@@ -0,0 +1,202 @@
+"""
+Tests for muxterm wiring into FastAPI lifespan (task-8).
+
+Verifies:
+- ttyd imports removed, muxterm imports present
+- _muxterm_secret auto-generates when MUXTERM_SECRET env var is empty
+- lifespan calls start_muxterm on boot with correct args
+- lifespan calls stop_muxterm on shutdown
+- FileNotFoundError from start_muxterm is caught gracefully
+- Generic Exception from start_muxterm is caught gracefully
+"""
+
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+# ---------------------------------------------------------------------------
+# Import structure tests
+# ---------------------------------------------------------------------------
+
+
+def test_main_does_not_import_ttyd_names():
+    """main.py must not export ttyd names (kill_orphan_ttyd, spawn_ttyd, etc.)."""
+    import muxplex.main as main_mod
+
+    assert not hasattr(main_mod, "kill_orphan_ttyd")
+    assert not hasattr(main_mod, "spawn_ttyd")
+    assert not hasattr(main_mod, "kill_ttyd")
+
+
+def test_main_imports_muxterm_functions():
+    """main.py must import start_muxterm and stop_muxterm from muxterm module."""
+    import muxplex.main as main_mod
+
+    assert hasattr(main_mod, "start_muxterm")
+    assert hasattr(main_mod, "stop_muxterm")
+
+
+# ---------------------------------------------------------------------------
+# Secret auto-generation test
+# ---------------------------------------------------------------------------
+
+
+def test_muxterm_secret_auto_generates_when_env_empty():
+    """_muxterm_secret must be a non-empty hex string when MUXTERM_SECRET env is unset."""
+    import muxplex.main as main_mod
+
+    secret = main_mod._muxterm_secret
+    assert isinstance(secret, str)
+    assert len(secret) > 0, "_muxterm_secret must auto-generate when env var is empty"
+
+
+# ---------------------------------------------------------------------------
+# Lifespan wiring tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def patch_startup_deps(tmp_path, monkeypatch):
+    """Redirect state files, mock muxterm and poll loop for clean lifespan tests."""
+    # Redirect state files
+    tmp_state_dir = tmp_path / "state"
+    tmp_state_path = tmp_state_dir / "state.json"
+    monkeypatch.setattr("muxplex.state.STATE_DIR", tmp_state_dir)
+    monkeypatch.setattr("muxplex.state.STATE_PATH", tmp_state_path)
+
+    # Replace _poll_loop with no-op
+    async def noop_poll_loop() -> None:
+        pass
+
+    monkeypatch.setattr("muxplex.main._poll_loop", noop_poll_loop)
+
+
+@pytest.fixture(autouse=True)
+def reset_federation_cache():
+    """Clear _federation_cache before and after each test."""
+    import muxplex.main as main_mod
+
+    main_mod._federation_cache.clear()
+    yield
+    main_mod._federation_cache.clear()
+
+
+def test_lifespan_calls_start_muxterm_on_boot(monkeypatch):
+    """Lifespan must call start_muxterm(secret=..., port=MUXTERM_PORT) on startup."""
+    from muxplex.main import app, MUXTERM_PORT
+
+    mock_start = AsyncMock()
+    mock_stop = AsyncMock()
+    monkeypatch.setattr("muxplex.main.start_muxterm", mock_start)
+    monkeypatch.setattr("muxplex.main.stop_muxterm", mock_stop)
+    monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password")
+
+    # Redirect state files
+    import tempfile
+    import pathlib
+
+    tmp = pathlib.Path(tempfile.mkdtemp())
+    monkeypatch.setattr("muxplex.state.STATE_DIR", tmp / "state")
+    monkeypatch.setattr("muxplex.state.STATE_PATH", tmp / "state" / "state.json")
+
+    async def noop_poll_loop() -> None:
+        pass
+
+    monkeypatch.setattr("muxplex.main._poll_loop", noop_poll_loop)
+
+    with TestClient(app):
+        pass  # lifespan runs on enter
+
+    mock_start.assert_called_once()
+    call_kwargs = mock_start.call_args
+    # start_muxterm should be called with secret= and port=
+    assert "secret" in call_kwargs.kwargs or len(call_kwargs.args) >= 1
+    assert call_kwargs.kwargs.get("port") == MUXTERM_PORT or (
+        len(call_kwargs.args) >= 2 and call_kwargs.args[1] == MUXTERM_PORT
+    )
+
+
+def test_lifespan_calls_stop_muxterm_on_shutdown(monkeypatch):
+    """Lifespan must call stop_muxterm() during shutdown."""
+    from muxplex.main import app
+
+    mock_start = AsyncMock()
+    mock_stop = AsyncMock()
+    monkeypatch.setattr("muxplex.main.start_muxterm", mock_start)
+    monkeypatch.setattr("muxplex.main.stop_muxterm", mock_stop)
+    monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password")
+
+    import tempfile
+    import pathlib
+
+    tmp = pathlib.Path(tempfile.mkdtemp())
+    monkeypatch.setattr("muxplex.state.STATE_DIR", tmp / "state")
+    monkeypatch.setattr("muxplex.state.STATE_PATH", tmp / "state" / "state.json")
+
+    async def noop_poll_loop() -> None:
+        pass
+
+    monkeypatch.setattr("muxplex.main._poll_loop", noop_poll_loop)
+
+    with TestClient(app):
+        pass  # lifespan runs on enter, shutdown on exit
+
+    mock_stop.assert_called_once()
+
+
+def test_lifespan_handles_file_not_found_error(monkeypatch):
+    """Lifespan must catch FileNotFoundError from start_muxterm gracefully."""
+    from muxplex.main import app
+
+    mock_start = AsyncMock(side_effect=FileNotFoundError("muxterm not found"))
+    mock_stop = AsyncMock()
+    monkeypatch.setattr("muxplex.main.start_muxterm", mock_start)
+    monkeypatch.setattr("muxplex.main.stop_muxterm", mock_stop)
+    monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password")
+
+    import tempfile
+    import pathlib
+
+    tmp = pathlib.Path(tempfile.mkdtemp())
+    monkeypatch.setattr("muxplex.state.STATE_DIR", tmp / "state")
+    monkeypatch.setattr("muxplex.state.STATE_PATH", tmp / "state" / "state.json")
+
+    async def noop_poll_loop() -> None:
+        pass
+
+    monkeypatch.setattr("muxplex.main._poll_loop", noop_poll_loop)
+
+    # Should NOT raise - lifespan catches FileNotFoundError
+    with TestClient(app) as c:
+        resp = c.get("/health")
+        assert resp.status_code == 200
+
+
+def test_lifespan_handles_generic_exception(monkeypatch):
+    """Lifespan must catch generic Exception from start_muxterm gracefully."""
+    from muxplex.main import app
+
+    mock_start = AsyncMock(side_effect=RuntimeError("something went wrong"))
+    mock_stop = AsyncMock()
+    monkeypatch.setattr("muxplex.main.start_muxterm", mock_start)
+    monkeypatch.setattr("muxplex.main.stop_muxterm", mock_stop)
+    monkeypatch.setenv("MUXPLEX_PASSWORD", "test-password")
+
+    import tempfile
+    import pathlib
+
+    tmp = pathlib.Path(tempfile.mkdtemp())
+    monkeypatch.setattr("muxplex.state.STATE_DIR", tmp / "state")
+    monkeypatch.setattr("muxplex.state.STATE_PATH", tmp / "state" / "state.json")
+
+    async def noop_poll_loop() -> None:
+        pass
+
+    monkeypatch.setattr("muxplex.main._poll_loop", noop_poll_loop)
+
+    # Should NOT raise - lifespan catches generic Exception
+    with TestClient(app) as c:
+        resp = c.get("/health")
+        assert resp.status_code == 200