forked from genewildish/Mainline
David Gwilliam
8d066edcca
Move internal imports in run_pipeline_mode() to module level to support proper mocking in integration tests. This enables more effective testing of the app's initialization and pipeline setup. Also simplifies the test suite to focus on key integration points. Changes: - Moved effects_plugins, DisplayRegistry, PerformanceMonitor, fetch functions, and pipeline adapters to module-level imports - Removed duplicate imports from run_pipeline_mode() - Simplified test_app.py to focus on core functionality All manual tests still pass (border-test preset works correctly).
234 lines
9.3 KiB
Python
234 lines
9.3 KiB
Python
"""
|
|
Integration tests for engine/app.py - pipeline orchestration.
|
|
|
|
Tests the main entry point and pipeline mode initialization,
|
|
including preset loading, display creation, and stage setup.
|
|
"""
|
|
|
|
import sys
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from engine.app import main, run_pipeline_mode
|
|
from engine.display import DisplayRegistry, NullDisplay
|
|
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.run_pipeline_mode") as mock_run:
|
|
sys.argv = ["mainline.py"]
|
|
main()
|
|
mock_run.assert_called_once_with("demo")
|
|
|
|
def test_main_uses_preset_from_config(self):
|
|
"""main() uses PRESET from config if set."""
|
|
with (
|
|
patch("engine.app.config") as mock_config,
|
|
patch("engine.app.run_pipeline_mode") as mock_run,
|
|
):
|
|
mock_config.PRESET = "border-test"
|
|
sys.argv = ["mainline.py"]
|
|
main()
|
|
mock_run.assert_called_once_with("border-test")
|
|
|
|
def test_main_exits_on_unknown_preset(self):
|
|
"""main() exits with error message for unknown preset."""
|
|
sys.argv = ["mainline.py"]
|
|
with (
|
|
patch("engine.app.config") as mock_config,
|
|
patch("engine.app.list_presets", return_value=["demo", "poetry"]),
|
|
pytest.raises(SystemExit) as exc_info,
|
|
):
|
|
mock_config.PRESET = "nonexistent"
|
|
main()
|
|
assert exc_info.value.code == 1
|
|
|
|
|
|
class TestRunPipelineMode:
|
|
"""Test run_pipeline_mode() pipeline setup and execution."""
|
|
|
|
def setup_method(self):
|
|
"""Setup for each test."""
|
|
DisplayRegistry._backends = {}
|
|
DisplayRegistry._initialized = False
|
|
DisplayRegistry.register("null", NullDisplay)
|
|
|
|
def test_run_pipeline_mode_loads_preset(self):
|
|
"""run_pipeline_mode() loads the specified preset."""
|
|
preset = get_preset("demo")
|
|
assert preset is not None
|
|
assert preset.name == "demo"
|
|
|
|
def test_run_pipeline_mode_exits_on_unknown_preset(self):
|
|
"""run_pipeline_mode() exits if preset not found."""
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
run_pipeline_mode("nonexistent-preset")
|
|
assert exc_info.value.code == 1
|
|
|
|
def test_run_pipeline_mode_fetches_content_for_headlines(self):
|
|
"""run_pipeline_mode() fetches content for headlines preset."""
|
|
with (
|
|
patch("engine.app.load_cache", return_value=None),
|
|
patch(
|
|
"engine.app.fetch_all", return_value=(["item1", "item2"], None, None)
|
|
),
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.effects_plugins"),
|
|
patch("engine.app.Pipeline") as mock_pipeline_class,
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline_class.return_value = mock_pipeline
|
|
|
|
# Will timeout after first iteration due to KeyboardInterrupt
|
|
with pytest.raises((StopIteration, AttributeError)):
|
|
run_pipeline_mode("demo")
|
|
|
|
def test_run_pipeline_mode_handles_empty_source(self):
|
|
"""run_pipeline_mode() handles empty source without fetching."""
|
|
with (
|
|
patch("engine.app.fetch_all") as mock_fetch,
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.effects_plugins"),
|
|
patch("engine.app.Pipeline") as mock_pipeline_class,
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline_class.return_value = mock_pipeline
|
|
|
|
try:
|
|
run_pipeline_mode("border-test")
|
|
except (StopIteration, AttributeError):
|
|
pass
|
|
# Should NOT call fetch_all for empty source
|
|
mock_fetch.assert_not_called()
|
|
|
|
def test_run_pipeline_mode_uses_cached_content(self):
|
|
"""run_pipeline_mode() uses cached content if available."""
|
|
cached_items = ["cached1", "cached2"]
|
|
with (
|
|
patch("engine.app.load_cache", return_value=cached_items),
|
|
patch("engine.app.fetch_all") as mock_fetch,
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.effects_plugins"),
|
|
patch("engine.app.Pipeline") as mock_pipeline_class,
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline_class.return_value = mock_pipeline
|
|
|
|
try:
|
|
run_pipeline_mode("demo")
|
|
except (StopIteration, AttributeError):
|
|
pass
|
|
# Should NOT call fetch_all when cache exists
|
|
mock_fetch.assert_not_called()
|
|
|
|
def test_run_pipeline_mode_fetches_poetry_for_poetry_preset(self):
|
|
"""run_pipeline_mode() fetches poetry for poetry source."""
|
|
with (
|
|
patch("engine.app.load_cache", return_value=None),
|
|
patch("engine.app.fetch_poetry", return_value=(["poem1"], None, None)),
|
|
patch("engine.app.fetch_all") as mock_fetch_all,
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.effects_plugins"),
|
|
patch("engine.app.Pipeline") as mock_pipeline_class,
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline_class.return_value = mock_pipeline
|
|
|
|
try:
|
|
run_pipeline_mode("poetry")
|
|
except (StopIteration, AttributeError):
|
|
pass
|
|
# Should NOT call fetch_all for poetry preset
|
|
mock_fetch_all.assert_not_called()
|
|
|
|
def test_run_pipeline_mode_exits_when_no_content(self):
|
|
"""run_pipeline_mode() exits if no content available."""
|
|
with (
|
|
patch("engine.app.load_cache", return_value=None),
|
|
patch("engine.app.fetch_all", return_value=([], None, None)),
|
|
patch("engine.app.effects_plugins"),
|
|
pytest.raises(SystemExit) as exc_info,
|
|
):
|
|
run_pipeline_mode("demo")
|
|
assert exc_info.value.code == 1
|
|
|
|
def test_run_pipeline_mode_display_flag_overrides_preset(self):
|
|
"""run_pipeline_mode() uses CLI --display flag over preset."""
|
|
sys.argv = ["mainline.py", "--preset", "border-test", "--display", "null"]
|
|
with (
|
|
patch("engine.app.load_cache", return_value=None),
|
|
patch("engine.app.fetch_all", return_value=(["item"], None, None)),
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.effects_plugins"),
|
|
patch("engine.app.Pipeline") as mock_pipeline_class,
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline_class.return_value = mock_pipeline
|
|
|
|
try:
|
|
run_pipeline_mode("border-test")
|
|
except (StopIteration, AttributeError):
|
|
pass
|
|
# Should create display with null, not terminal
|
|
mock_create.assert_called_with("null")
|
|
|
|
def test_run_pipeline_mode_discovers_effects_plugins(self):
|
|
"""run_pipeline_mode() discovers available effect plugins."""
|
|
with (
|
|
patch("engine.app.effects_plugins") as mock_effects,
|
|
patch("engine.app.load_cache", return_value=["item"]),
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.Pipeline") as mock_pipeline_class,
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
mock_pipeline = MagicMock()
|
|
mock_pipeline_class.return_value = mock_pipeline
|
|
|
|
try:
|
|
run_pipeline_mode("demo")
|
|
except (StopIteration, AttributeError):
|
|
pass
|
|
# Should call discover_plugins
|
|
mock_effects.discover_plugins.assert_called_once()
|
|
|
|
def test_run_pipeline_mode_initializes_display(self):
|
|
"""run_pipeline_mode() initializes display with dimensions."""
|
|
with (
|
|
patch("engine.app.load_cache", return_value=["item"]),
|
|
patch("engine.app.DisplayRegistry.create") as mock_create,
|
|
patch("engine.app.effects_plugins"),
|
|
patch("engine.app.Pipeline"),
|
|
patch("engine.app.time.sleep"),
|
|
):
|
|
mock_display = MagicMock()
|
|
mock_create.return_value = mock_display
|
|
|
|
try:
|
|
run_pipeline_mode("demo")
|
|
except (StopIteration, AttributeError):
|
|
pass
|
|
# Display should be initialized with dimensions
|
|
mock_display.init.assert_called()
|