diff --git a/fastloop/utils.py b/fastloop/utils.py index 2eef549..ccf5522 100644 --- a/fastloop/utils.py +++ b/fastloop/utils.py @@ -58,13 +58,16 @@ def infer_application_path(app_instance: Any, fallback_var: str = "app") -> str Application path string like "module.path:app" or None if cannot be determined """ # (1) Introspect app if it has an 'app' attribute (e.g., FastLoop wrapping FastAPI) - app = getattr(app_instance, "app", None) or app_instance + # Note: We don't fall back to app_instance here because app_instance.__module__ + # would be "fastloop.fastloop" (where the class is defined), not the user's module. + # If there's no 'app' attribute, we fall through to argv-based inference. + app = getattr(app_instance, "app", None) if app is not None and getattr(app, "__module__", None): mod_name = app.__module__ try: mod = importlib.import_module(mod_name) for name, val in vars(mod).items(): - if val is app or val is app_instance: + if val is app: return f"{mod_name}:{name}" # app object found but variable name not recoverable — use fallback var return f"{mod_name}:{fallback_var}" diff --git a/tests/test_app_initialization.py b/tests/test_app_initialization.py new file mode 100644 index 0000000..5ebf23f --- /dev/null +++ b/tests/test_app_initialization.py @@ -0,0 +1,169 @@ +""" +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