1+ from importlib import reload
2+ from types import ModuleType
3+
4+ import pytest
5+ from fastapi import FastAPI
6+ from app .utils .observability import Instrumentator # type: ignore
7+
8+ # List of (module path, attribute name) tuples for each micro-service FastAPI app
9+ SERVICE_MODULES = [
10+ ("app.main_user" , "service_app" ),
11+ ("app.main_comment" , "service_app" ),
12+ ("app.main_video" , "service_app" ),
13+ ("app.main_reco" , "service_app" ),
14+ ("app.main_search" , "service_app" ),
15+ ("app.main_moderation" , "service_app" ),
16+ ]
17+
18+
19+ @pytest .mark .parametrize ("module_path, attr" , SERVICE_MODULES )
20+ def test_metrics_route_present_once (module_path : str , attr : str ):
21+ """Ensure each micro-service exposes exactly one /metrics endpoint."""
22+
23+ # Reload observability module so that its module-level flags are reset for each service
24+ reload (__import__ ("app.utils.observability" , fromlist = ["*" ]))
25+
26+ mod : ModuleType = __import__ (module_path , fromlist = [attr ])
27+ app : FastAPI = getattr (mod , attr )
28+
29+ if Instrumentator is None : # pragma: no cover
30+ pytest .skip ("prometheus_fastapi_instrumentator not installed; skipping metrics route test" )
31+
32+ metrics_routes = [r for r in app .routes if getattr (r , "path" , None ) == "/metrics" ]
33+
34+ assert len (metrics_routes ) == 1 , (
35+ f"Expected exactly one /metrics route in { module_path } , found { len (metrics_routes )} "
36+ )
0 commit comments