forked from genewildish/Mainline
David Gwilliam
ad8513f2f6
- Patch instead of - Add missing patches for and in background threads - Prevent network I/O during tests
216 lines
8.7 KiB
Python
216 lines
8.7 KiB
Python
"""
|
|
Integration tests for engine/app.py - pipeline orchestration.
|
|
|
|
Tests the main entry point and pipeline mode initialization.
|
|
"""
|
|
|
|
import sys
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from engine.app import main, run_pipeline_mode
|
|
from engine.pipeline import get_preset
|
|
|
|
|
|
class TestMain:
|
|
"""Test main() entry point."""
|
|
|
|
def test_main_calls_run_pipeline_mode_with_default_preset(self):
|
|
"""main() runs default preset (demo) when no args provided."""
|
|
with patch("engine.app.main.run_pipeline_mode") as mock_run:
|
|
sys.argv = ["mainline.py"]
|
|
main()
|
|
mock_run.assert_called_once_with("demo")
|
|
|
|
def test_main_calls_run_pipeline_mode_with_config_preset(self):
|
|
"""main() uses PRESET from config if set."""
|
|
with (
|
|
patch("engine.config.PIPELINE_DIAGRAM", False),
|
|
patch("engine.config.PRESET", "demo"),
|
|
patch("engine.config.PIPELINE_MODE", False),
|
|
patch("engine.app.main.run_pipeline_mode") as mock_run,
|
|
):
|
|
sys.argv = ["mainline.py"]
|
|
main()
|
|
mock_run.assert_called_once_with("demo")
|
|
|
|
def test_main_exits_on_unknown_preset(self):
|
|
"""main() exits with error for unknown preset."""
|
|
with (
|
|
patch("engine.config.PIPELINE_DIAGRAM", False),
|
|
patch("engine.config.PRESET", "nonexistent"),
|
|
patch("engine.config.PIPELINE_MODE", False),
|
|
patch("engine.pipeline.list_presets", return_value=["demo", "poetry"]),
|
|
):
|
|
sys.argv = ["mainline.py"]
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
main()
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
class TestRunPipelineMode:
|
|
"""Test run_pipeline_mode() initialization."""
|
|
|
|
def test_run_pipeline_mode_loads_valid_preset(self):
|
|
"""run_pipeline_mode() loads a valid preset."""
|
|
preset = get_preset("demo")
|
|
assert preset is not None
|
|
assert preset.name == "demo"
|
|
assert preset.source == "headlines"
|
|
|
|
def test_run_pipeline_mode_exits_on_invalid_preset(self):
|
|
"""run_pipeline_mode() exits if preset not found."""
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
run_pipeline_mode("invalid-preset-xyz")
|
|
assert exc_info.value.code == 1
|
|
|
|
def test_run_pipeline_mode_exits_when_no_content_available(self):
|
|
"""run_pipeline_mode() exits if no content can be fetched."""
|
|
with (
|
|
patch("engine.app.pipeline_runner.load_cache", return_value=None),
|
|
patch("engine.app.pipeline_runner.fetch_all_fast", return_value=[]),
|
|
patch(
|
|
"engine.app.pipeline_runner.fetch_all", return_value=([], None, None)
|
|
), # Mock background thread
|
|
patch("engine.app.pipeline_runner.save_cache"), # Prevent disk I/O
|
|
patch("engine.effects.plugins.discover_plugins"),
|
|
pytest.raises(SystemExit) as exc_info,
|
|
):
|
|
run_pipeline_mode("demo")
|
|
assert exc_info.value.code == 1
|
|
|
|
def test_run_pipeline_mode_uses_cache_over_fetch(self):
|
|
"""run_pipeline_mode() uses cached content if available."""
|
|
cached = ["cached_item"]
|
|
with (
|
|
patch(
|
|
"engine.app.pipeline_runner.load_cache", return_value=cached
|
|
) as mock_load,
|
|
patch("engine.app.pipeline_runner.fetch_all") as mock_fetch,
|
|
patch("engine.app.pipeline_runner.fetch_all_fast"),
|
|
patch("engine.app.pipeline_runner.DisplayRegistry.create") as mock_create,
|
|
):
|
|
mock_display = Mock()
|
|
mock_display.init = Mock()
|
|
mock_display.get_dimensions = Mock(return_value=(80, 24))
|
|
mock_display.is_quit_requested = Mock(return_value=True)
|
|
mock_display.clear_quit_request = Mock()
|
|
mock_display.show = Mock()
|
|
mock_display.cleanup = Mock()
|
|
mock_create.return_value = mock_display
|
|
|
|
try:
|
|
run_pipeline_mode("demo")
|
|
except (KeyboardInterrupt, SystemExit):
|
|
pass
|
|
|
|
# Verify fetch_all was NOT called (cache was used)
|
|
mock_fetch.assert_not_called()
|
|
mock_load.assert_called_once()
|
|
|
|
def test_run_pipeline_mode_creates_display(self):
|
|
"""run_pipeline_mode() creates a display backend."""
|
|
with (
|
|
patch("engine.app.pipeline_runner.load_cache", return_value=["item"]),
|
|
patch("engine.app.pipeline_runner.fetch_all_fast", return_value=[]),
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
):
|
|
mock_display = Mock()
|
|
mock_display.init = Mock()
|
|
mock_display.get_dimensions = Mock(return_value=(80, 24))
|
|
mock_display.is_quit_requested = Mock(return_value=True)
|
|
mock_display.clear_quit_request = Mock()
|
|
mock_display.show = Mock()
|
|
mock_display.cleanup = Mock()
|
|
mock_create.return_value = mock_display
|
|
|
|
try:
|
|
run_pipeline_mode("demo-base")
|
|
except (KeyboardInterrupt, SystemExit):
|
|
pass
|
|
|
|
# Verify display was created with 'terminal' (preset display)
|
|
mock_create.assert_called_once_with("terminal")
|
|
|
|
def test_run_pipeline_mode_respects_display_cli_flag(self):
|
|
"""run_pipeline_mode() uses --display CLI flag if provided."""
|
|
sys.argv = ["mainline.py", "--display", "websocket"]
|
|
|
|
with (
|
|
patch("engine.app.pipeline_runner.load_cache", return_value=["item"]),
|
|
patch("engine.app.pipeline_runner.fetch_all_fast", return_value=[]),
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
):
|
|
mock_display = Mock()
|
|
mock_display.init = Mock()
|
|
mock_display.get_dimensions = Mock(return_value=(80, 24))
|
|
mock_display.is_quit_requested = Mock(return_value=True)
|
|
mock_display.clear_quit_request = Mock()
|
|
mock_display.show = Mock()
|
|
mock_display.cleanup = Mock()
|
|
mock_create.return_value = mock_display
|
|
|
|
try:
|
|
run_pipeline_mode("demo")
|
|
except (KeyboardInterrupt, SystemExit):
|
|
pass
|
|
|
|
# Verify display was created with CLI override
|
|
mock_create.assert_called_once_with("websocket")
|
|
|
|
def test_run_pipeline_mode_fetches_poetry_for_poetry_source(self):
|
|
"""run_pipeline_mode() fetches poetry for poetry preset."""
|
|
with (
|
|
patch("engine.app.pipeline_runner.load_cache", return_value=None),
|
|
patch(
|
|
"engine.app.pipeline_runner.fetch_poetry",
|
|
return_value=(["poem"], None, None),
|
|
) as mock_fetch_poetry,
|
|
patch("engine.app.pipeline_runner.fetch_all") as mock_fetch_all,
|
|
patch("engine.app.pipeline_runner.fetch_all_fast", return_value=[]),
|
|
patch("engine.app.pipeline_runner.DisplayRegistry.create") as mock_create,
|
|
):
|
|
mock_display = Mock()
|
|
mock_display.init = Mock()
|
|
mock_display.get_dimensions = Mock(return_value=(80, 24))
|
|
mock_display.is_quit_requested = Mock(return_value=True)
|
|
mock_display.clear_quit_request = Mock()
|
|
mock_display.show = Mock()
|
|
mock_display.cleanup = Mock()
|
|
mock_create.return_value = mock_display
|
|
|
|
try:
|
|
run_pipeline_mode("poetry")
|
|
except (KeyboardInterrupt, SystemExit):
|
|
pass
|
|
|
|
# Verify fetch_poetry was called, not fetch_all
|
|
mock_fetch_poetry.assert_called_once()
|
|
mock_fetch_all.assert_not_called()
|
|
|
|
def test_run_pipeline_mode_discovers_effect_plugins(self):
|
|
"""run_pipeline_mode() discovers available effect plugins."""
|
|
with (
|
|
patch("engine.app.pipeline_runner.load_cache", return_value=["item"]),
|
|
patch("engine.app.pipeline_runner.fetch_all_fast", return_value=[]),
|
|
patch("engine.effects.plugins.discover_plugins") as mock_discover,
|
|
patch("engine.app.pipeline_runner.DisplayRegistry.create") as mock_create,
|
|
):
|
|
mock_display = Mock()
|
|
mock_display.init = Mock()
|
|
mock_display.get_dimensions = Mock(return_value=(80, 24))
|
|
mock_display.is_quit_requested = Mock(return_value=True)
|
|
mock_display.clear_quit_request = Mock()
|
|
mock_display.show = Mock()
|
|
mock_display.cleanup = Mock()
|
|
mock_create.return_value = mock_display
|
|
|
|
try:
|
|
run_pipeline_mode("demo")
|
|
except (KeyboardInterrupt, SystemExit):
|
|
pass
|
|
|
|
# Verify effects_plugins.discover_plugins was called
|
|
mock_discover.assert_called_once()
|