fixes tests

Author: pcrespov

services/web/server/src/simcore_service_webserver/login/handlers_change.py

@@ -49,14 +49,6 @@ class ResetPasswordBody(InputSchema):
     email: str
 
 
-def _get_request_context(request: web.Request) -> dict[str, str]:
-    return {
-        "request.remote": f"{request.remote}",
-        "request.method": f"{request.method}",
-        "request.path": f"{request.path}",
-    }
-
-
 @routes.post(f"/{API_VTAG}/auth/reset-password", name="initiate_reset_password")
 @global_rate_limit_route(number_of_requests=10, interval_seconds=HOUR)
 async def initiate_reset_password(request: web.Request):
@@ -80,27 +72,50 @@ async def initiate_reset_password(request: web.Request):
 
     request_body = await parse_request_body_as(ResetPasswordBody, request)
 
+    _error_msg_prefix = "Password reset initiated"
+
+    def _get_error_context(
+        user=None,
+    ) -> dict[str, str]:
+        ctx = {
+            "user_email": request_body.email,
+            "product_name": product.name,
+            "request.remote": f"{request.remote}",
+            "request.method": f"{request.method}",
+            "request.path": f"{request.path}",
+        }
+
+        if user:
+            ctx.update(
+                {
+                    "user_email": request_body.email,
+                    "user_id": user["id"],
+                    "user_status": user["status"],
+                    "user_role": user["role"],
+                }
+            )
+        return ctx
+
     # NOTE: Always same response: never want to confirm or deny the existence of an account
     # with a given email or username.
     initiated_response = flash_response(
         MSG_EMAIL_SENT.format(email=request_body.email), "INFO"
     )
+
     # CHECK user exists
     user = await db.get_user({"email": request_body.email})
     if not user:
         _logger.warning(
             **create_troubleshotting_log_kwargs(
-                "Password reset initiated for non-existent email. Ignoring request.",
+                "{_error_msg_prefix} for non-existent email. Ignoring request.",
                 error=Exception("No user found with this email"),
-                error_context={
-                    "user_email": request_body.email,
-                    "product_name": product.name,
-                    **_get_request_context(request),
-                },
+                error_context=_get_error_context(),
             )
         )
         return initiated_response
 
+    assert user["email"] == request_body.email  # nosec
+
     # CHECK user state
     try:
         validate_user_status(user=dict(user), support_email=product.support_email)
@@ -109,20 +124,14 @@ async def initiate_reset_password(request: web.Request):
         # do not want to forward but rather log due to the special rules in this entrypoint
         _logger.warning(
             **create_troubleshotting_log_kwargs(
-                "Password reset initiated for invalid user. Ignoring request.",
+                f"{_error_msg_prefix} for invalid user. Ignoring request.",
                 error=err,
-                error_context={
-                    "user_id": user["id"],
-                    "user": user,
-                    "product_name": product.name,
-                    **_get_request_context(request),
-                },
+                error_context=_get_error_context(user),
             )
         )
         return initiated_response
 
     assert user["status"] == ACTIVE  # nosec
-    assert user["email"] == request_body.email  # nosec
     assert isinstance(user["id"], int)  # nosec
 
     # CHECK access to product
@@ -131,14 +140,9 @@ async def initiate_reset_password(request: web.Request):
     ):
         _logger.warning(
             **create_troubleshotting_log_kwargs(
-                "Password reset initiated for a user with NO access to this product. Ignoring request.",
+                f"{_error_msg_prefix} for a user with NO access to this product. Ignoring request.",
                 error=Exception("User cannot access this product"),
-                error_context={
-                    "user_id": user["id"],
-                    "user": user,
-                    "product_name": product.name,
-                    **_get_request_context(request),
-                },
+                error_context=_get_error_context(user),
             )
         )
         return initiated_response
@@ -169,11 +173,7 @@ async def initiate_reset_password(request: web.Request):
             **create_troubleshotting_log_kwargs(
                 "Unable to send email",
                 error=err,
-                error_context={
-                    "product_name": product.name,
-                    "user_id": user["id"],
-                    **_get_request_context(request),
-                },
+                error_context=_get_error_context(user),
             )
         )
         raise web.HTTPServiceUnavailable(reason=MSG_CANT_SEND_MAIL) from err

services/web/server/tests/unit/with_dbs/03/login/test_login_reset_password.py

@@ -138,14 +138,16 @@ async def test_unknown_email(
 
     # email is not sent
     out, _ = capsys.readouterr()
-    assert not parse_test_marks(out)
+    assert not parse_test_marks(out), "Expected no email to be sent"
 
     # Check logger warning
+    logged_warnings = [
+        record.message for record in caplog.records if record.levelname == "WARNING"
+    ]
+
     assert any(
-        record.levelname == "WARNING"
-        and record.message.startswith("Password reset initiated")
-        for record in caplog.records
-    )
+        message.startswith("Password reset initiated") for message in logged_warnings
+    ), f"Missing warning in {logged_warnings}"
 
 
 @pytest.mark.parametrize(
@@ -158,6 +160,7 @@ async def test_unknown_email(
 async def test_blocked_user(
     client: TestClient,
     capsys: pytest.CaptureFixture,
+    caplog: pytest.LogCaptureFixture,
     user_status: UserStatus,
     expected_msg: str,
 ):
@@ -175,12 +178,24 @@ async def test_blocked_user(
     assert response.url.path == reset_url.path
     await assert_status(response, status.HTTP_200_OK, MSG_EMAIL_SENT.format(**user))
 
+    # email is not sent
     out, _ = capsys.readouterr()
+    assert not parse_test_marks(out), "Expected no email to be sent"
+
     # expected_msg contains {support_email} at the end of the string
-    assert expected_msg[:-20] in parse_test_marks(out)["reason"]
+    logged_warnings = [
+        record.message for record in caplog.records if record.levelname == "WARNING"
+    ]
+
+    assert any(
+        message.startswith("Password reset initiated") and expected_msg[:50] in message
+        for message in logged_warnings
+    ), f"Missing warning in {logged_warnings}"
 
 
-async def test_inactive_user(client: TestClient, capsys: pytest.CaptureFixture):
+async def test_inactive_user(
+    client: TestClient, capsys: pytest.CaptureFixture, caplog: pytest.LogCaptureFixture
+):
     assert client.app
     reset_url = client.app.router["initiate_reset_password"].url_for()
 
@@ -197,8 +212,19 @@ async def test_inactive_user(client: TestClient, capsys: pytest.CaptureFixture):
     assert response.url.path == reset_url.path
     await assert_status(response, status.HTTP_200_OK, MSG_EMAIL_SENT.format(**user))
 
+    # email is not sent
     out, _ = capsys.readouterr()
-    assert parse_test_marks(out)["reason"] == MSG_ACTIVATION_REQUIRED
+    assert not parse_test_marks(out), "Expected no email to be sent"
+
+    logged_warnings = [
+        record.message for record in caplog.records if record.levelname == "WARNING"
+    ]
+
+    assert any(
+        message.startswith("Password reset initiated")
+        and MSG_ACTIVATION_REQUIRED[:20] in message
+        for message in logged_warnings
+    ), f"Missing warning in {logged_warnings}"
 
 
 async def test_too_often(