"""Tests for router container.""" import sys import pytest from fastapi import APIRouter, FastAPI from fastapi_route_loader.container import RouterContainer from fastapi_route_loader.events import ( RouterEventType, RouterLoadedEvent, RouterUnloadedEvent, RouterUpdatedEvent, ) class TestRouterContainer: """Test router container functionality.""" def test_add_router(self): container = RouterContainer() router = APIRouter() container.add_router("test_router", router) assert container.has_router("test_router") assert container.get_router("test_router") is router assert len(container) == 1 def test_add_router_duplicate_raises_error(self): container = RouterContainer() router = APIRouter() container.add_router("test_router", router) with pytest.raises(ValueError, match="Router 'test_router' already exists"): container.add_router("test_router", router) def test_add_router_dispatches_event(self): container = RouterContainer() router = APIRouter() events_received = [] def handler(event): events_received.append(event) container.on(RouterEventType.LOADED, handler) container.add_router("test_router", router, {"key": "value"}) assert len(events_received) == 1 event = events_received[0] assert isinstance(event, RouterLoadedEvent) assert event.router_name == "test_router" assert event.router is router assert event.metadata == {"key": "value"} def test_remove_router(self): container = RouterContainer() router = APIRouter() container.add_router("test_router", router) removed_router = container.remove_router("test_router") assert removed_router is router assert not container.has_router("test_router") assert len(container) == 0 def test_remove_router_not_found_raises_error(self): container = RouterContainer() with pytest.raises(KeyError, match="Router 'nonexistent' not found"): container.remove_router("nonexistent") def test_remove_router_dispatches_event(self): container = RouterContainer() router = APIRouter() events_received = [] def handler(event): events_received.append(event) container.add_router("test_router", router) container.on(RouterEventType.UNLOADED, handler) container.remove_router("test_router", {"reason": "cleanup"}) assert len(events_received) == 1 event = events_received[0] assert isinstance(event, RouterUnloadedEvent) assert event.router_name == "test_router" assert event.router is router assert event.metadata == {"reason": "cleanup"} def test_update_router(self): container = RouterContainer() old_router = APIRouter() new_router = APIRouter() container.add_router("test_router", old_router) container.update_router("test_router", new_router) assert container.get_router("test_router") is new_router def test_update_router_not_found_raises_error(self): container = RouterContainer() router = APIRouter() with pytest.raises(KeyError, match="Router 'nonexistent' not found"): container.update_router("nonexistent", router) def test_update_router_dispatches_event(self): container = RouterContainer() old_router = APIRouter() new_router = APIRouter() events_received = [] def handler(event): events_received.append(event) container.add_router("test_router", old_router) container.on(RouterEventType.UPDATED, handler) container.update_router("test_router", new_router, {"version": "2.0"}) assert len(events_received) == 1 event = events_received[0] assert isinstance(event, RouterUpdatedEvent) assert event.router_name == "test_router" assert event.router is new_router assert event.old_router is old_router assert event.metadata == {"version": "2.0"} def test_get_router_not_found_raises_error(self): container = RouterContainer() with pytest.raises(KeyError, match="Router 'nonexistent' not found"): container.get_router("nonexistent") def test_exclude_routers(self): container = RouterContainer() router1 = APIRouter() router2 = APIRouter() router3 = APIRouter() container.add_router("router1", router1) container.add_router("router2", router2) container.add_router("router3", router3) container.exclude("router2") assert container.is_active("router1") assert not container.is_active("router2") assert container.is_active("router3") active_routers = container.get_active_routers() assert "router1" in active_routers assert "router2" not in active_routers assert "router3" in active_routers def test_include_routers(self): container = RouterContainer() router1 = APIRouter() router2 = APIRouter() router3 = APIRouter() container.add_router("router1", router1) container.add_router("router2", router2) container.add_router("router3", router3) container.include("router1", "router3") assert container.is_active("router1") assert not container.is_active("router2") assert container.is_active("router3") active_routers = container.get_active_routers() assert "router1" in active_routers assert "router2" not in active_routers assert "router3" in active_routers def test_clear_filters(self): container = RouterContainer() router1 = APIRouter() router2 = APIRouter() container.add_router("router1", router1) container.add_router("router2", router2) container.exclude("router1") assert not container.is_active("router1") container.clear_filters() assert container.is_active("router1") assert container.is_active("router2") def test_get_all_routers(self): container = RouterContainer() router1 = APIRouter() router2 = APIRouter() container.add_router("router1", router1) container.add_router("router2", router2) container.exclude("router1") all_routers = container.get_all_routers() assert len(all_routers) == 2 assert "router1" in all_routers assert "router2" in all_routers def test_load_from_module(self, tmp_path): module_content = ''' from fastapi import APIRouter test_router = APIRouter() ''' module_file = tmp_path / "test_module.py" module_file.write_text(module_content) sys.path.insert(0, str(tmp_path)) try: container = RouterContainer() container.load_from_module("test_module") assert container.has_router("test_router") finally: sys.path.remove(str(tmp_path)) if "test_module" in sys.modules: del sys.modules["test_module"] def test_load_from_directory(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\nuser_router = APIRouter()" ) sys.path.insert(0, str(tmp_path)) try: container = RouterContainer() container.load_from_directory(str(routers_dir), "routers") assert container.has_router("users.user_router") finally: sys.path.remove(str(tmp_path)) for mod in list(sys.modules.keys()): if mod.startswith("routers"): del sys.modules[mod] def test_register_to_app(self): container = RouterContainer() router1 = APIRouter() router2 = APIRouter() container.add_router("router1", router1) container.add_router("router2", router2) container.exclude("router2") app = FastAPI() container.register_to_app(app) assert len(app.routes) > 0 def test_register_to_app_with_prefix(self): container = RouterContainer() router = APIRouter() router.get("/")(lambda: {"message": "test"}) container.add_router("router", router) app = FastAPI() container.register_to_app(app, prefix="/api") route_paths = [route.path for route in app.routes] assert any("/api" in path for path in route_paths) def test_event_subscription(self): container = RouterContainer() router = APIRouter() events_received = [] def handler(event): events_received.append(event) container.on(None, handler) container.add_router("test_router", router) container.remove_router("test_router") assert len(events_received) == 2 def test_event_unsubscription(self): container = RouterContainer() router = APIRouter() events_received = [] def handler(event): events_received.append(event) container.on(RouterEventType.LOADED, handler) container.off(RouterEventType.LOADED, handler) container.add_router("test_router", router) assert len(events_received) == 0 def test_clear_handlers(self): container = RouterContainer() router = APIRouter() events_received = [] def handler(event): events_received.append(event) container.on(RouterEventType.LOADED, handler) container.clear_handlers() container.add_router("test_router", router) assert len(events_received) == 0 def test_container_len(self): container = RouterContainer() assert len(container) == 0 container.add_router("router1", APIRouter()) assert len(container) == 1 container.add_router("router2", APIRouter()) assert len(container) == 2 container.remove_router("router1") assert len(container) == 1 def test_container_contains(self): container = RouterContainer() router = APIRouter() container.add_router("test_router", router) assert "test_router" in container assert "nonexistent" not in container def test_container_iter(self): container = RouterContainer() container.add_router("router1", APIRouter()) container.add_router("router2", APIRouter()) container.add_router("router3", APIRouter()) router_names = list(container) assert "router1" in router_names assert "router2" in router_names assert "router3" in router_names assert len(router_names) == 3