Dev

src/app/api/v1/endpoints/__init__.py

@@ -2,7 +2,7 @@
 
 from . import email, logs, push, user_config, user_history
 
-router = APIRouter(tags=["Notifications"])
+router = APIRouter()
 
 router.include_router(push.router)
 router.include_router(email.router)

src/app/api/v1/endpoints/logs.py

@@ -8,7 +8,7 @@
 from app.services.logs_service import LogsService
 
 logger = setup_logger(__name__)
-router = APIRouter()
+router = APIRouter(tags=["Logs"])
 
 
 @router.get(

src/app/api/v1/endpoints/push.py

@@ -8,7 +8,7 @@
 
 logger = setup_logger(__name__)
 
-router = APIRouter()
+router = APIRouter(tags=["Send"])
 
 
 @router.post(

src/app/api/v1/endpoints/user_config.py

@@ -14,7 +14,7 @@
 
 logger = setup_logger(__name__)
 
-router = APIRouter(prefix="/user")
+router = APIRouter(prefix="/user", tags=["User"])
 
 
 @router.put(

src/app/api/v1/endpoints/user_history.py

@@ -14,7 +14,7 @@
 
 logger = setup_logger(__name__)
 
-router = APIRouter(prefix="/user")
+router = APIRouter(prefix="/user", tags=["user"])
 
 
 @router.get(

src/app/middleware/jw_admin_auth.py

@@ -28,7 +28,7 @@ def jw_admin_middleware(request: Request) -> Optional[bool]:
             token_string, settings.ADMIN_SERVICE_SECRET, algorithms=["HS256"]
         )
         is_admin = payload.get("isAdmin")
-        if isinstance(is_admin, bool):
+        if isinstance(is_admin, bool) and is_admin:
             return is_admin
         else:
             logger.warning("Missing or invalid isAdmin in JWT claims")

tests/unit/test_jw_admin_auth.py

@@ -0,0 +1,97 @@
+from datetime import datetime, timedelta, timezone
+
+import jwt
+import pytest
+from fastapi import FastAPI, Request
+from fastapi.testclient import TestClient
+
+from app.errors.errors import no_user_id
+from app.middleware.jw_admin_auth import jw_admin_middleware
+
+
+# Mock de settings
+class Settings:
+    ADMIN_SERVICE_SECRET = "adminsecret"
+
+
+settings = Settings()
+
+
+# Fixture para la aplicación de prueba
+@pytest.fixture
+def test_admin_app():
+    app = FastAPI()
+
+    @app.get("/admin-test")
+    async def admin_test_route(request: Request):
+        auth_result = jw_admin_middleware(request)
+
+        if auth_result is None and hasattr(request.state, "invalid_jwt"):
+            return no_user_id(request)
+
+        if auth_result is True:
+            return {"status": "success", "admin": True}
+        elif auth_result is False:
+            return {"status": "success", "admin": False}
+
+        return {"status": "success", "message": "no auth provided"}
+
+    return app
+
+
+# Fixture para el admin de prueba
+@pytest.fixture
+def admin_client(test_admin_app):
+    return TestClient(test_admin_app)
+
+
+# Función para crear tokens de prueba
+def create_admin_token(secret: str, is_admin: bool, exp_delta=1) -> str:
+    payload = {
+        "isAdmin": is_admin,
+        "exp": now_utc() + timedelta(hours=exp_delta),
+    }
+    return jwt.encode(payload, secret, algorithm="HS256")
+
+
+def test_admin_no_auth_header(admin_client):
+    response = admin_client.get("/admin-test")
+    assert response.status_code == 200
+    assert response.json() == {"status": "success", "message": "no auth provided"}
+
+
+def test_admin_invalid_auth_format(admin_client):
+    response = admin_client.get(
+        "/admin-test", headers={"Authorization": "InvalidHeader"}
+    )
+    assert response.status_code == 401
+    assert response.json()["title"] == "Invalid token"
+
+
+def test_admin_token_missing_is_admin(admin_client):
+    token = jwt.encode(
+        {"exp": now_utc() + timedelta(hours=1)},
+        settings.ADMIN_SERVICE_SECRET,
+        algorithm="HS256",
+    )
+    response = admin_client.get(
+        "/admin-test", headers={"Authorization": f"Bearer {token}"}
+    )
+    assert response.status_code == 401
+    assert response.json()["title"] == "Invalid token"
+
+
+def test_admin_expired_token(admin_client):
+    expired_token = create_admin_token(
+        settings.ADMIN_SERVICE_SECRET, is_admin=True, exp_delta=-1
+    )
+    response = admin_client.get(
+        "/admin-test", headers={"Authorization": f"Bearer {expired_token}"}
+    )
+    assert response.status_code == 401
+    assert response.json()["title"] == "Invalid token"
+
+
+# Función para obtener el datetime actual con timezone UTC
+def now_utc():
+    return datetime.now(timezone.utc)

tests/unit/test_jw_auth.py

@@ -53,7 +53,6 @@ def create_test_token(
     return jwt.encode(payload, secret, algorithm="HS256")
 
 
-# Tests actualizados
 def test_no_auth_header(client):
     response = client.get("/test")
     assert response.status_code == 200

tests/unit/test_logs_endpoint.py

@@ -0,0 +1,108 @@
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+from fastapi import status
+from httpx import ASGITransport, AsyncClient
+
+from app.main import app
+from app.middleware.jw_admin_auth import jw_admin_middleware
+
+
+@pytest_asyncio.fixture
+async def client():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(
+        transport=transport, base_url="http://notifications-test"
+    ) as ac:
+        yield ac
+
+
+def get_logs_url():
+    return "/notifications/logs"
+
+
+@pytest.mark.asyncio
+async def test_get_logs_success(client: AsyncClient):
+    logs_response = {
+        "total": 2,
+        "logs": [
+            {"_id": "log1", "notif_type": "email", "notif_status": "sent"},
+            {"_id": "log2", "notif_type": "push", "notif_status": "failed"},
+        ],
+    }
+
+    app.dependency_overrides[jw_admin_middleware] = lambda: {
+        "user_id": "admin_123",
+        "role": "admin",
+    }
+
+    with patch(
+        "app.api.v1.endpoints.logs.LogsService.get_logs_service", new_callable=AsyncMock
+    ) as mock_get_logs:
+        mock_get_logs.return_value = logs_response
+
+        response = await client.get(
+            "/notifications/logs?skip=0&limit=10&date_order=desc"
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json() == logs_response
+
+    app.dependency_overrides.pop(jw_admin_middleware)
+
+
+@pytest.mark.asyncio
+async def test_get_logs_unauthorized(client: AsyncClient):
+    app.dependency_overrides[jw_admin_middleware] = lambda: None
+
+    response = await client.get("/notifications/logs")
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+    assert "detail" in response.json()
+
+    app.dependency_overrides.pop(jw_admin_middleware)
+
+
+@pytest.mark.asyncio
+async def test_get_logs_service_http_error(client: AsyncClient):
+    from fastapi import HTTPException
+
+    app.dependency_overrides[jw_admin_middleware] = lambda: {
+        "user_id": "admin_123",
+        "role": "admin",
+    }
+
+    with patch(
+        "app.api.v1.endpoints.logs.LogsService.get_logs_service", new_callable=AsyncMock
+    ) as mock_get_logs:
+        mock_get_logs.side_effect = HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Logs not found"
+        )
+
+        response = await client.get("/notifications/logs")
+
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        assert response.json()["title"] == "Logs Error"
+
+    app.dependency_overrides.pop(jw_admin_middleware)
+
+
+@pytest.mark.asyncio
+async def test_get_logs_service_unexpected_error(client: AsyncClient):
+    app.dependency_overrides[jw_admin_middleware] = lambda: {
+        "user_id": "admin_123",
+        "role": "admin",
+    }
+
+    with patch(
+        "app.api.v1.endpoints.logs.LogsService.get_logs_service", new_callable=AsyncMock
+    ) as mock_get_logs:
+        mock_get_logs.side_effect = Exception("DB exploded")
+
+        response = await client.get("/notifications/logs")
+
+        assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+        assert response.json()["title"] == "Internal Server Error"
+
+    app.dependency_overrides.pop(jw_admin_middleware)