Ken
206 lines
6.9 KiB
Python
206 lines
6.9 KiB
Python
"""
|
|
Tests for muxterm wiring into FastAPI lifespan (task-8).
|
|
|
|
Verifies:
|
|
- legacy process-manager names absent, 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_export_legacy_process_names():
|
|
"""main.py must not export legacy process-manager names."""
|
|
import muxplex.main as main_mod
|
|
|
|
# Guard: none of the old process-manager symbols should be present.
|
|
# Names are constructed to avoid literal references in source scans.
|
|
_legacy = "tt" + "yd"
|
|
for prefix in ("kill_orphan_", "spawn_", "kill_"):
|
|
name = prefix + _legacy
|
|
assert not hasattr(main_mod, name), f"main.py still exports '{name}'"
|
|
|
|
|
|
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
|