262 lines
7.2 KiB
Python
262 lines
7.2 KiB
Python
"""Integration tests for the complete library."""
|
|
|
|
import sys
|
|
|
|
from fastapi import APIRouter, FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fastapi_route_loader import RouterContainer, RouterEventType
|
|
|
|
|
|
class TestIntegration:
|
|
"""Test complete integration scenarios."""
|
|
|
|
def test_full_workflow(self, tmp_path):
|
|
routers_dir = tmp_path / "routers"
|
|
routers_dir.mkdir()
|
|
|
|
(routers_dir / "__init__.py").write_text("")
|
|
(routers_dir / "users.py").write_text('''
|
|
from fastapi import APIRouter
|
|
|
|
users_router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
@users_router.get("/")
|
|
def list_users():
|
|
return {"users": []}
|
|
|
|
@users_router.get("/{user_id}")
|
|
def get_user(user_id: int):
|
|
return {"user_id": user_id}
|
|
''')
|
|
|
|
(routers_dir / "posts.py").write_text('''
|
|
from fastapi import APIRouter
|
|
|
|
posts_router = APIRouter(prefix="/posts", tags=["posts"])
|
|
|
|
@posts_router.get("/")
|
|
def list_posts():
|
|
return {"posts": []}
|
|
''')
|
|
|
|
sys.path.insert(0, str(tmp_path))
|
|
try:
|
|
container = RouterContainer()
|
|
container.load_from_directory(str(routers_dir), "routers")
|
|
|
|
assert container.has_router("users.users_router")
|
|
assert container.has_router("posts.posts_router")
|
|
|
|
app = FastAPI()
|
|
container.register_to_app(app)
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/users/")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"users": []}
|
|
|
|
response = client.get("/users/123")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"user_id": 123}
|
|
|
|
response = client.get("/posts/")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"posts": []}
|
|
finally:
|
|
sys.path.remove(str(tmp_path))
|
|
for mod in list(sys.modules.keys()):
|
|
if mod.startswith("routers"):
|
|
del sys.modules[mod]
|
|
|
|
def test_filtering_workflow(self, tmp_path):
|
|
routers_dir = tmp_path / "routers"
|
|
routers_dir.mkdir()
|
|
|
|
(routers_dir / "__init__.py").write_text("")
|
|
(routers_dir / "public.py").write_text('''
|
|
from fastapi import APIRouter
|
|
|
|
public_router = APIRouter(prefix="/public")
|
|
|
|
@public_router.get("/")
|
|
def public_endpoint():
|
|
return {"message": "public"}
|
|
''')
|
|
|
|
(routers_dir / "admin.py").write_text('''
|
|
from fastapi import APIRouter
|
|
|
|
admin_router = APIRouter(prefix="/admin")
|
|
|
|
@admin_router.get("/")
|
|
def admin_endpoint():
|
|
return {"message": "admin"}
|
|
''')
|
|
|
|
sys.path.insert(0, str(tmp_path))
|
|
try:
|
|
container = RouterContainer()
|
|
container.load_from_directory(str(routers_dir), "routers")
|
|
|
|
container.exclude("admin.admin_router")
|
|
|
|
app = FastAPI()
|
|
container.register_to_app(app)
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/public/")
|
|
assert response.status_code == 200
|
|
|
|
response = client.get("/admin/")
|
|
assert response.status_code == 404
|
|
finally:
|
|
sys.path.remove(str(tmp_path))
|
|
for mod in list(sys.modules.keys()):
|
|
if mod.startswith("routers"):
|
|
del sys.modules[mod]
|
|
|
|
def test_event_tracking_workflow(self, tmp_path):
|
|
routers_dir = tmp_path / "routers"
|
|
routers_dir.mkdir()
|
|
|
|
(routers_dir / "__init__.py").write_text("")
|
|
(routers_dir / "api.py").write_text('''
|
|
from fastapi import APIRouter
|
|
|
|
api_router = APIRouter()
|
|
''')
|
|
|
|
sys.path.insert(0, str(tmp_path))
|
|
try:
|
|
container = RouterContainer()
|
|
events_log = []
|
|
|
|
def event_logger(event):
|
|
events_log.append({
|
|
"type": event.event_type.value,
|
|
"router": event.router_name,
|
|
})
|
|
|
|
container.on(None, event_logger)
|
|
|
|
container.load_from_directory(str(routers_dir), "routers")
|
|
|
|
assert len(events_log) == 1
|
|
assert events_log[0]["type"] == "loaded"
|
|
assert events_log[0]["router"] == "api.api_router"
|
|
|
|
new_router = APIRouter()
|
|
container.update_router("api.api_router", new_router)
|
|
|
|
assert len(events_log) == 2
|
|
assert events_log[1]["type"] == "updated"
|
|
|
|
container.remove_router("api.api_router")
|
|
|
|
assert len(events_log) == 3
|
|
assert events_log[2]["type"] == "unloaded"
|
|
finally:
|
|
sys.path.remove(str(tmp_path))
|
|
for mod in list(sys.modules.keys()):
|
|
if mod.startswith("routers"):
|
|
del sys.modules[mod]
|
|
|
|
def test_manual_router_management(self):
|
|
container = RouterContainer()
|
|
|
|
users_router = APIRouter(prefix="/users")
|
|
|
|
@users_router.get("/")
|
|
def list_users():
|
|
return {"users": []}
|
|
|
|
posts_router = APIRouter(prefix="/posts")
|
|
|
|
@posts_router.get("/")
|
|
def list_posts():
|
|
return {"posts": []}
|
|
|
|
container.add_router("users", users_router)
|
|
container.add_router("posts", posts_router)
|
|
|
|
app = FastAPI()
|
|
container.register_to_app(app)
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/users/")
|
|
assert response.status_code == 200
|
|
|
|
response = client.get("/posts/")
|
|
assert response.status_code == 200
|
|
|
|
def test_include_only_specific_routers(self):
|
|
container = RouterContainer()
|
|
|
|
router1 = APIRouter(prefix="/api1")
|
|
router2 = APIRouter(prefix="/api2")
|
|
router3 = APIRouter(prefix="/api3")
|
|
|
|
@router1.get("/")
|
|
def api1():
|
|
return {"api": 1}
|
|
|
|
@router2.get("/")
|
|
def api2():
|
|
return {"api": 2}
|
|
|
|
@router3.get("/")
|
|
def api3():
|
|
return {"api": 3}
|
|
|
|
container.add_router("api1", router1)
|
|
container.add_router("api2", router2)
|
|
container.add_router("api3", router3)
|
|
|
|
container.include("api1", "api3")
|
|
|
|
app = FastAPI()
|
|
container.register_to_app(app)
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/api1/")
|
|
assert response.status_code == 200
|
|
|
|
response = client.get("/api2/")
|
|
assert response.status_code == 404
|
|
|
|
response = client.get("/api3/")
|
|
assert response.status_code == 200
|
|
|
|
def test_event_handlers_for_specific_types(self):
|
|
container = RouterContainer()
|
|
|
|
loaded_count = [0]
|
|
unloaded_count = [0]
|
|
updated_count = [0]
|
|
|
|
def on_loaded(event):
|
|
loaded_count[0] += 1
|
|
|
|
def on_unloaded(event):
|
|
unloaded_count[0] += 1
|
|
|
|
def on_updated(event):
|
|
updated_count[0] += 1
|
|
|
|
container.on(RouterEventType.LOADED, on_loaded)
|
|
container.on(RouterEventType.UNLOADED, on_unloaded)
|
|
container.on(RouterEventType.UPDATED, on_updated)
|
|
|
|
router1 = APIRouter()
|
|
router2 = APIRouter()
|
|
|
|
container.add_router("router1", router1)
|
|
assert loaded_count[0] == 1
|
|
|
|
container.update_router("router1", router2)
|
|
assert updated_count[0] == 1
|
|
|
|
container.remove_router("router1")
|
|
assert unloaded_count[0] == 1
|