test_app_initialization.py

"""
Regression tests for FastLoop app initialization.

These tests ensure that:
1. The application path inference works correctly for hypercorn hot reload
2. FastLoop apps with registered loops initialize properly

Regression for: When infer_application_path incorrectly returns "fastloop.fastloop:app",
hypercorn fails with:
    NoAppError: Cannot load application from 'fastloop.fastloop:app', application not found.
"""

import sys
import tempfile
from pathlib import Path
from unittest import mock

from fastloop import FastLoop
from fastloop.context import LoopContext
from fastloop.loop import LoopEvent
from fastloop.utils import infer_application_path

# --- Event Types for Testing ---


class QueryEvent(LoopEvent):
    type: str = "query"
    message: str


class ResponseEvent(LoopEvent):
    type: str = "response"
    reply: str


# --- Test Classes ---


class TestInferApplicationPath:
    """
    Regression tests for infer_application_path.

    The bug: When a FastLoop instance has no 'app' attribute, the function
    was falling back to app_instance itself, returning "fastloop.fastloop:app"
    which doesn't exist.
    """

    def test_does_not_return_fastloop_module_path(self):
        """FastLoop instances should NOT resolve to the fastloop package."""
        app = FastLoop(name="test-app")
        result = infer_application_path(app)

        if result is not None:
            assert not result.startswith("fastloop."), (
                f"infer_application_path returned '{result}' pointing to fastloop package. "
                "This causes: NoAppError: Cannot load application from 'fastloop.fastloop:app'"
            )

    def test_falls_back_to_argv_inference(self):
        """When no module var found, should use argv-based inference."""
        app = FastLoop(name="test-app")

        with tempfile.TemporaryDirectory() as tmpdir:
            script_path = Path(tmpdir) / "myapp" / "main.py"
            script_path.parent.mkdir(parents=True)
            script_path.touch()

            with (
                mock.patch.object(sys, "argv", [str(script_path)]),
                mock.patch.object(sys, "path", [tmpdir, *sys.path]),
            ):
                result = infer_application_path(app)

                if result is not None:
                    assert "myapp.main" in result
                    assert ":app" in result


class TestFastLoopWithLoops:
    """Tests for FastLoop app with registered loops."""

    def test_app_with_loop_decorator(self):
        """
        Test that a FastLoop app with a @loop decorator initializes correctly
        and doesn't cause import issues.
        """
        app = FastLoop(name="test-chat-app")
        app.register_events([QueryEvent, ResponseEvent])

        async def on_start(context: LoopContext):
            await context.set("initialized", True)

        @app.loop("chat", start_event=QueryEvent, on_start=on_start)
        async def chat_loop(context: LoopContext):
            msg = await context.wait_for(QueryEvent, raise_on_timeout=False, timeout=1)
            if msg is None:
                return
            await context.emit(ResponseEvent(reply=f"Echo: {msg.message}"))

        # Verify loop was registered
        assert "chat" in app._loop_metadata
        assert app._loop_metadata["chat"]["func"] == chat_loop
        assert app._loop_metadata["chat"]["start_event"] == "query"

        # Verify events were registered
        assert "query" in app._event_types
        assert "response" in app._event_types

        # Verify infer_application_path doesn't break
        result = infer_application_path(app)
        if result is not None:
            assert not result.startswith("fastloop.")

    def test_app_with_multiple_loops(self):
        """Test app with multiple loop definitions."""
        app = FastLoop(name="multi-loop-app")
        app.register_events([QueryEvent, ResponseEvent])

        @app.loop("loop-a", start_event=QueryEvent)
        async def loop_a(context: LoopContext):
            pass

        @app.loop("loop-b", start_event=ResponseEvent)
        async def loop_b(context: LoopContext):
            pass

        assert "loop-a" in app._loop_metadata
        assert "loop-b" in app._loop_metadata
        assert len(app._loop_metadata) == 2

    def test_app_routes_registered(self):
        """Verify that loop decorator registers the expected API routes."""
        app = FastLoop(name="route-test-app")
        app.register_events([QueryEvent])

        @app.loop("myloop", start_event=QueryEvent)
        async def my_loop(context: LoopContext):
            pass

        # Check routes were added
        route_paths = [route.path for route in app.routes]
        assert "/myloop" in route_paths
        assert "/myloop/{loop_id}" in route_paths
        assert "/myloop/{loop_id}/stop" in route_paths
        assert "/myloop/{loop_id}/pause" in route_paths


class TestAppConfiguration:
    """Tests for FastLoop configuration."""

    def test_debug_mode_defaults_to_false(self):
        """Debug mode should be off by default."""
        app = FastLoop(name="test-app")
        assert app.config_manager.get("debugMode", False) is False

    def test_custom_config_applied(self):
        """Custom config should be merged with defaults."""
        app = FastLoop(
            name="test-app",
            config={"port": 9000, "debugMode": True},
        )
        assert app.config_manager.get("port") == 9000
        assert app.config_manager.get("debugMode") is True

    def test_fastloop_has_no_app_attribute(self):
        """FastLoop should not have an 'app' attribute (it IS the app)."""
        app = FastLoop(name="test-app")
        # FastLoop is a FastAPI subclass, not a wrapper
        assert getattr(app, "app", None) is None