From f0752ba6d2a4781b7ddec80b871ee2516ecac58b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 26 Nov 2025 15:40:36 +0000 Subject: [PATCH 1/2] Fix: Improve app path inference and add regression tests Co-authored-by: luke --- fastloop/utils.py | 7 +- tests/test_app_initialization.py | 178 +++++++++++++++++++++++++++++++ 2 files changed, 183 insertions(+), 2 deletions(-) create mode 100644 tests/test_app_initialization.py 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..eb914a1 --- /dev/null +++ b/tests/test_app_initialization.py @@ -0,0 +1,178 @@ +""" +Regression tests for FastLoop app initialization. + +These tests ensure that the application path inference works correctly +and that hypercorn can properly load the application for hot reload. + +Related to bug: https://github.com/user/fastloop/issues/XXX +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 +import textwrap +from pathlib import Path +from unittest import mock + +from fastloop import FastLoop +from fastloop.utils import infer_application_path + + +class TestInferApplicationPath: + """Tests for the infer_application_path utility function.""" + + def test_does_not_return_fastloop_module_path(self): + """ + Regression test: infer_application_path should NOT return a path + pointing to the fastloop package itself. + + When a user creates `app = FastLoop(...)`, the app's __module__ is + "fastloop.fastloop" (where the class is defined). But hypercorn needs + to load the app from the USER's module, not fastloop's. + """ + app = FastLoop(name="test-app") + result = infer_application_path(app) + + # The result should NOT be "fastloop.fastloop:app" or similar + # because there's no `app` variable in fastloop.fastloop module + if result is not None: + assert not result.startswith("fastloop."), ( + f"infer_application_path returned '{result}' which points to the " + "fastloop package. This would cause hypercorn to fail with " + "'NoAppError: Cannot load application from fastloop.fastloop:app'" + ) + + def test_returns_none_when_app_not_in_module_vars(self): + """ + When the FastLoop instance isn't assigned to a module-level variable + that can be found, infer_application_path should return None or + fall back to argv-based inference. + """ + # Create an app that's not assigned to any module variable + app = FastLoop(name="test-app") + + # Mock sys.argv to simulate running a script + with mock.patch.object(sys, "argv", ["test_script.py"]): + result = infer_application_path(app) + + # Should either return None or a valid path (not fastloop.fastloop:*) + if result is not None: + assert not result.startswith("fastloop.") + + def test_with_app_attribute_none(self): + """ + Test that if the app_instance has an 'app' attribute that is None, + we fall through to argv-based inference. + """ + + class AppWrapper: + def __init__(self): + self.app = None + + wrapper = AppWrapper() + + # When app attribute is None, should fall through to argv inference + result = infer_application_path(wrapper) + + # Should not return fastloop path since we don't have a valid app + if result is not None: + assert not result.startswith("fastloop.") + + def test_argv_based_inference(self): + """ + Test that argv-based inference works as a fallback. + """ + app = FastLoop(name="test-app") + + # Create a temporary Python file to simulate a script + with tempfile.TemporaryDirectory() as tmpdir: + script_path = Path(tmpdir) / "myproject" / "main.py" + script_path.parent.mkdir(parents=True) + script_path.touch() + + # Mock sys.argv and sys.path + with ( + mock.patch.object(sys, "argv", [str(script_path)]), + mock.patch.object(sys, "path", [tmpdir, *sys.path]), + ): + result = infer_application_path(app) + + # Should return something like "myproject.main:app" + if result is not None: + assert "myproject.main" in result + assert ":app" in result + + +class TestFastLoopRunConfiguration: + """Tests for FastLoop.run() configuration.""" + + def test_application_path_not_set_in_non_debug_mode(self): + """ + In non-debug mode, application_path should not be set, + and we should use asyncio.run directly. + """ + app = FastLoop(name="test-app") + + # The run method should use asyncio.run when not in debug mode + # We can't easily test this without actually running the server, + # but we verify the config manager defaults + assert app.config_manager.get("debugMode", False) is False + + def test_fastloop_instance_has_no_app_attribute(self): + """ + Verify that FastLoop doesn't have an 'app' attribute that would + confuse the introspection logic. + """ + app = FastLoop(name="test-app") + + # FastLoop is a FastAPI subclass, not a wrapper with an 'app' attribute + assert not hasattr(app, "app") or getattr(app, "app", None) is None + + +class TestHypercornCompatibility: + """Tests for hypercorn compatibility.""" + + def test_infer_application_path_returns_valid_format(self): + """ + When infer_application_path returns a value, it should be in + the format "module.path:variable_name". + """ + # Create a temporary module with an app + with tempfile.TemporaryDirectory() as tmpdir: + module_dir = Path(tmpdir) / "testpkg" + module_dir.mkdir() + (module_dir / "__init__.py").touch() + + app_file = module_dir / "application.py" + app_file.write_text( + textwrap.dedent(""" + from fastloop import FastLoop + app = FastLoop(name="test") + """) + ) + + # Add to path and import + sys.path.insert(0, tmpdir) + try: + import testpkg.application + + result = infer_application_path(testpkg.application.app) + + if result is not None: + # Should be in format "module:var" + assert ":" in result, f"Result '{result}' should contain ':'" + module_part, var_part = result.rsplit(":", 1) + assert len(module_part) > 0 + assert len(var_part) > 0 + + # Should not point to fastloop + assert not module_part.startswith("fastloop") + finally: + sys.path.remove(tmpdir) + # Clean up imported module + if "testpkg.application" in sys.modules: + del sys.modules["testpkg.application"] + if "testpkg" in sys.modules: + del sys.modules["testpkg"] From 1bc7cbf3ff6e37e1472f728d796d70debb15c270 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 26 Nov 2025 15:42:37 +0000 Subject: [PATCH 2/2] Refactor tests and improve app initialization logic Co-authored-by: luke --- tests/test_app_initialization.py | 233 +++++++++++++++---------------- 1 file changed, 112 insertions(+), 121 deletions(-) diff --git a/tests/test_app_initialization.py b/tests/test_app_initialization.py index eb914a1..5ebf23f 100644 --- a/tests/test_app_initialization.py +++ b/tests/test_app_initialization.py @@ -1,178 +1,169 @@ """ Regression tests for FastLoop app initialization. -These tests ensure that the application path inference works correctly -and that hypercorn can properly load the application for hot reload. +These tests ensure that: +1. The application path inference works correctly for hypercorn hot reload +2. FastLoop apps with registered loops initialize properly -Related to bug: https://github.com/user/fastloop/issues/XXX -When infer_application_path incorrectly returns "fastloop.fastloop:app", +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 -import textwrap 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 TestInferApplicationPath: - """Tests for the infer_application_path utility function.""" - - def test_does_not_return_fastloop_module_path(self): - """ - Regression test: infer_application_path should NOT return a path - pointing to the fastloop package itself. - When a user creates `app = FastLoop(...)`, the app's __module__ is - "fastloop.fastloop" (where the class is defined). But hypercorn needs - to load the app from the USER's module, not fastloop's. - """ - app = FastLoop(name="test-app") - result = infer_application_path(app) +class QueryEvent(LoopEvent): + type: str = "query" + message: str - # The result should NOT be "fastloop.fastloop:app" or similar - # because there's no `app` variable in fastloop.fastloop module - if result is not None: - assert not result.startswith("fastloop."), ( - f"infer_application_path returned '{result}' which points to the " - "fastloop package. This would cause hypercorn to fail with " - "'NoAppError: Cannot load application from fastloop.fastloop:app'" - ) - def test_returns_none_when_app_not_in_module_vars(self): - """ - When the FastLoop instance isn't assigned to a module-level variable - that can be found, infer_application_path should return None or - fall back to argv-based inference. - """ - # Create an app that's not assigned to any module variable - app = FastLoop(name="test-app") +class ResponseEvent(LoopEvent): + type: str = "response" + reply: str - # Mock sys.argv to simulate running a script - with mock.patch.object(sys, "argv", ["test_script.py"]): - result = infer_application_path(app) - # Should either return None or a valid path (not fastloop.fastloop:*) - if result is not None: - assert not result.startswith("fastloop.") +# --- Test Classes --- - def test_with_app_attribute_none(self): - """ - Test that if the app_instance has an 'app' attribute that is None, - we fall through to argv-based inference. - """ - class AppWrapper: - def __init__(self): - self.app = None +class TestInferApplicationPath: + """ + Regression tests for infer_application_path. - wrapper = AppWrapper() + 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. + """ - # When app attribute is None, should fall through to argv inference - result = infer_application_path(wrapper) + 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) - # Should not return fastloop path since we don't have a valid app if result is not None: - assert not result.startswith("fastloop.") + 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_argv_based_inference(self): - """ - Test that argv-based inference works as a fallback. - """ + def test_falls_back_to_argv_inference(self): + """When no module var found, should use argv-based inference.""" app = FastLoop(name="test-app") - # Create a temporary Python file to simulate a script with tempfile.TemporaryDirectory() as tmpdir: - script_path = Path(tmpdir) / "myproject" / "main.py" + script_path = Path(tmpdir) / "myapp" / "main.py" script_path.parent.mkdir(parents=True) script_path.touch() - # Mock sys.argv and sys.path with ( mock.patch.object(sys, "argv", [str(script_path)]), mock.patch.object(sys, "path", [tmpdir, *sys.path]), ): result = infer_application_path(app) - # Should return something like "myproject.main:app" if result is not None: - assert "myproject.main" in result + assert "myapp.main" in result assert ":app" in result -class TestFastLoopRunConfiguration: - """Tests for FastLoop.run() configuration.""" +class TestFastLoopWithLoops: + """Tests for FastLoop app with registered loops.""" - def test_application_path_not_set_in_non_debug_mode(self): + def test_app_with_loop_decorator(self): """ - In non-debug mode, application_path should not be set, - and we should use asyncio.run directly. + Test that a FastLoop app with a @loop decorator initializes correctly + and doesn't cause import issues. """ - app = FastLoop(name="test-app") + app = FastLoop(name="test-chat-app") + app.register_events([QueryEvent, ResponseEvent]) - # The run method should use asyncio.run when not in debug mode - # We can't easily test this without actually running the server, - # but we verify the config manager defaults - assert app.config_manager.get("debugMode", False) is False + async def on_start(context: LoopContext): + await context.set("initialized", True) - def test_fastloop_instance_has_no_app_attribute(self): - """ - Verify that FastLoop doesn't have an 'app' attribute that would - confuse the introspection logic. - """ - app = FastLoop(name="test-app") + @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}")) - # FastLoop is a FastAPI subclass, not a wrapper with an 'app' attribute - assert not hasattr(app, "app") or getattr(app, "app", None) is None + # 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 -class TestHypercornCompatibility: - """Tests for hypercorn compatibility.""" + # Verify infer_application_path doesn't break + result = infer_application_path(app) + if result is not None: + assert not result.startswith("fastloop.") - def test_infer_application_path_returns_valid_format(self): - """ - When infer_application_path returns a value, it should be in - the format "module.path:variable_name". - """ - # Create a temporary module with an app - with tempfile.TemporaryDirectory() as tmpdir: - module_dir = Path(tmpdir) / "testpkg" - module_dir.mkdir() - (module_dir / "__init__.py").touch() - - app_file = module_dir / "application.py" - app_file.write_text( - textwrap.dedent(""" - from fastloop import FastLoop - app = FastLoop(name="test") - """) - ) + def test_app_with_multiple_loops(self): + """Test app with multiple loop definitions.""" + app = FastLoop(name="multi-loop-app") + app.register_events([QueryEvent, ResponseEvent]) - # Add to path and import - sys.path.insert(0, tmpdir) - try: - import testpkg.application + @app.loop("loop-a", start_event=QueryEvent) + async def loop_a(context: LoopContext): + pass - result = infer_application_path(testpkg.application.app) + @app.loop("loop-b", start_event=ResponseEvent) + async def loop_b(context: LoopContext): + pass - if result is not None: - # Should be in format "module:var" - assert ":" in result, f"Result '{result}' should contain ':'" - module_part, var_part = result.rsplit(":", 1) - assert len(module_part) > 0 - assert len(var_part) > 0 - - # Should not point to fastloop - assert not module_part.startswith("fastloop") - finally: - sys.path.remove(tmpdir) - # Clean up imported module - if "testpkg.application" in sys.modules: - del sys.modules["testpkg.application"] - if "testpkg" in sys.modules: - del sys.modules["testpkg"] + 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