Skip to content

Commit 2063b69

Browse files
Fix fastloop initialization and add regression test (#16)
* Fix: Improve app path inference and add regression tests Co-authored-by: luke <luke@smartshare.io> * Refactor tests and improve app initialization logic Co-authored-by: luke <luke@smartshare.io> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
1 parent c8561df commit 2063b69

2 files changed

Lines changed: 174 additions & 2 deletions

File tree

fastloop/utils.py

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,13 +58,16 @@ def infer_application_path(app_instance: Any, fallback_var: str = "app") -> str
5858
Application path string like "module.path:app" or None if cannot be determined
5959
"""
6060
# (1) Introspect app if it has an 'app' attribute (e.g., FastLoop wrapping FastAPI)
61-
app = getattr(app_instance, "app", None) or app_instance
61+
# Note: We don't fall back to app_instance here because app_instance.__module__
62+
# would be "fastloop.fastloop" (where the class is defined), not the user's module.
63+
# If there's no 'app' attribute, we fall through to argv-based inference.
64+
app = getattr(app_instance, "app", None)
6265
if app is not None and getattr(app, "__module__", None):
6366
mod_name = app.__module__
6467
try:
6568
mod = importlib.import_module(mod_name)
6669
for name, val in vars(mod).items():
67-
if val is app or val is app_instance:
70+
if val is app:
6871
return f"{mod_name}:{name}"
6972
# app object found but variable name not recoverable — use fallback var
7073
return f"{mod_name}:{fallback_var}"

tests/test_app_initialization.py

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
"""
2+
Regression tests for FastLoop app initialization.
3+
4+
These tests ensure that:
5+
1. The application path inference works correctly for hypercorn hot reload
6+
2. FastLoop apps with registered loops initialize properly
7+
8+
Regression for: When infer_application_path incorrectly returns "fastloop.fastloop:app",
9+
hypercorn fails with:
10+
NoAppError: Cannot load application from 'fastloop.fastloop:app', application not found.
11+
"""
12+
13+
import sys
14+
import tempfile
15+
from pathlib import Path
16+
from unittest import mock
17+
18+
from fastloop import FastLoop
19+
from fastloop.context import LoopContext
20+
from fastloop.loop import LoopEvent
21+
from fastloop.utils import infer_application_path
22+
23+
# --- Event Types for Testing ---
24+
25+
26+
class QueryEvent(LoopEvent):
27+
type: str = "query"
28+
message: str
29+
30+
31+
class ResponseEvent(LoopEvent):
32+
type: str = "response"
33+
reply: str
34+
35+
36+
# --- Test Classes ---
37+
38+
39+
class TestInferApplicationPath:
40+
"""
41+
Regression tests for infer_application_path.
42+
43+
The bug: When a FastLoop instance has no 'app' attribute, the function
44+
was falling back to app_instance itself, returning "fastloop.fastloop:app"
45+
which doesn't exist.
46+
"""
47+
48+
def test_does_not_return_fastloop_module_path(self):
49+
"""FastLoop instances should NOT resolve to the fastloop package."""
50+
app = FastLoop(name="test-app")
51+
result = infer_application_path(app)
52+
53+
if result is not None:
54+
assert not result.startswith("fastloop."), (
55+
f"infer_application_path returned '{result}' pointing to fastloop package. "
56+
"This causes: NoAppError: Cannot load application from 'fastloop.fastloop:app'"
57+
)
58+
59+
def test_falls_back_to_argv_inference(self):
60+
"""When no module var found, should use argv-based inference."""
61+
app = FastLoop(name="test-app")
62+
63+
with tempfile.TemporaryDirectory() as tmpdir:
64+
script_path = Path(tmpdir) / "myapp" / "main.py"
65+
script_path.parent.mkdir(parents=True)
66+
script_path.touch()
67+
68+
with (
69+
mock.patch.object(sys, "argv", [str(script_path)]),
70+
mock.patch.object(sys, "path", [tmpdir, *sys.path]),
71+
):
72+
result = infer_application_path(app)
73+
74+
if result is not None:
75+
assert "myapp.main" in result
76+
assert ":app" in result
77+
78+
79+
class TestFastLoopWithLoops:
80+
"""Tests for FastLoop app with registered loops."""
81+
82+
def test_app_with_loop_decorator(self):
83+
"""
84+
Test that a FastLoop app with a @loop decorator initializes correctly
85+
and doesn't cause import issues.
86+
"""
87+
app = FastLoop(name="test-chat-app")
88+
app.register_events([QueryEvent, ResponseEvent])
89+
90+
async def on_start(context: LoopContext):
91+
await context.set("initialized", True)
92+
93+
@app.loop("chat", start_event=QueryEvent, on_start=on_start)
94+
async def chat_loop(context: LoopContext):
95+
msg = await context.wait_for(QueryEvent, raise_on_timeout=False, timeout=1)
96+
if msg is None:
97+
return
98+
await context.emit(ResponseEvent(reply=f"Echo: {msg.message}"))
99+
100+
# Verify loop was registered
101+
assert "chat" in app._loop_metadata
102+
assert app._loop_metadata["chat"]["func"] == chat_loop
103+
assert app._loop_metadata["chat"]["start_event"] == "query"
104+
105+
# Verify events were registered
106+
assert "query" in app._event_types
107+
assert "response" in app._event_types
108+
109+
# Verify infer_application_path doesn't break
110+
result = infer_application_path(app)
111+
if result is not None:
112+
assert not result.startswith("fastloop.")
113+
114+
def test_app_with_multiple_loops(self):
115+
"""Test app with multiple loop definitions."""
116+
app = FastLoop(name="multi-loop-app")
117+
app.register_events([QueryEvent, ResponseEvent])
118+
119+
@app.loop("loop-a", start_event=QueryEvent)
120+
async def loop_a(context: LoopContext):
121+
pass
122+
123+
@app.loop("loop-b", start_event=ResponseEvent)
124+
async def loop_b(context: LoopContext):
125+
pass
126+
127+
assert "loop-a" in app._loop_metadata
128+
assert "loop-b" in app._loop_metadata
129+
assert len(app._loop_metadata) == 2
130+
131+
def test_app_routes_registered(self):
132+
"""Verify that loop decorator registers the expected API routes."""
133+
app = FastLoop(name="route-test-app")
134+
app.register_events([QueryEvent])
135+
136+
@app.loop("myloop", start_event=QueryEvent)
137+
async def my_loop(context: LoopContext):
138+
pass
139+
140+
# Check routes were added
141+
route_paths = [route.path for route in app.routes]
142+
assert "/myloop" in route_paths
143+
assert "/myloop/{loop_id}" in route_paths
144+
assert "/myloop/{loop_id}/stop" in route_paths
145+
assert "/myloop/{loop_id}/pause" in route_paths
146+
147+
148+
class TestAppConfiguration:
149+
"""Tests for FastLoop configuration."""
150+
151+
def test_debug_mode_defaults_to_false(self):
152+
"""Debug mode should be off by default."""
153+
app = FastLoop(name="test-app")
154+
assert app.config_manager.get("debugMode", False) is False
155+
156+
def test_custom_config_applied(self):
157+
"""Custom config should be merged with defaults."""
158+
app = FastLoop(
159+
name="test-app",
160+
config={"port": 9000, "debugMode": True},
161+
)
162+
assert app.config_manager.get("port") == 9000
163+
assert app.config_manager.get("debugMode") is True
164+
165+
def test_fastloop_has_no_app_attribute(self):
166+
"""FastLoop should not have an 'app' attribute (it IS the app)."""
167+
app = FastLoop(name="test-app")
168+
# FastLoop is a FastAPI subclass, not a wrapper
169+
assert getattr(app, "app", None) is None

0 commit comments

Comments
 (0)