Improve tests

Author: pentest-gg

fastapi_users_db_dynamodb/__init__.py

@@ -182,7 +182,7 @@ async def get_by_oauth_account(
                         oauth_acc.user_id,
                         consistent_read=instant_update,
                     )
-                except self.user_table.DoesNotExist:  # type: ignore
+                except self.user_table.DoesNotExist:  # type: ignore # pragma: no cover
                     return None
         return None
 
@@ -231,10 +231,8 @@ async def delete(self, user: UP) -> None:
         try:
             await user.delete(condition=self.user_table.id.exists())  # type: ignore
         except DeleteError as e:
-            raise ValueError(  # pragma: no cover
-                "User account could not be deleted."
-            ) from e
-        except PutError as e:
+            raise ValueError("User account could not be deleted.") from e
+        except PutError as e:  # pragma: no cover
             if e.cause_response_code == "ConditionalCheckFailedException":
                 raise ValueError(
                     "User account could not be deleted because it does not exist."
@@ -282,7 +280,7 @@ async def update_oauth_account(
                 raise ValueError(
                     "OAuth account could not be updated because it does not exist."
                 ) from e
-            raise ValueError(
+            raise ValueError(  # pragma: no cover
                 "OAuth account could not be updated because the table does not exist."
             ) from e
 

fastapi_users_db_dynamodb/access_token.py

@@ -97,7 +97,7 @@ async def create(self, create_dict: dict[str, Any] | AP) -> AP:
                 raise ValueError(
                     "Access token could not be created because it already exists."
                 ) from e
-            raise ValueError(
+            raise ValueError(  # pragma: no cover
                 "Access token could not be created because the table does not exist."
             ) from e
         return token
@@ -116,7 +116,7 @@ async def update(self, access_token: AP, update_dict: dict[str, Any]) -> AP:
                 raise ValueError(
                     "Access token could not be updated because it does not exist."
                 ) from e
-            raise ValueError(
+            raise ValueError(  # pragma: no cover
                 "Access token could not be updated because the table does not exist."
             ) from e
 
@@ -127,10 +127,8 @@ async def delete(self, access_token: AP) -> None:
         try:
             await access_token.delete(condition=self.access_token_table.token.exists())  # type: ignore
         except DeleteError as e:
-            raise ValueError(  # pragma: no cover
-                "Access token could not be deleted."
-            ) from e
-        except PutError as e:
+            raise ValueError("Access token could not be deleted.") from e
+        except PutError as e:  # pragma: no cover
             if e.cause_response_code == "ConditionalCheckFailedException":
                 raise ValueError(
                     "Access token could not be deleted because it does not exist."

tests/test_access_token.py

@@ -110,6 +110,9 @@ async def test_queries(
 
     # Delete
     await dynamodb_access_token_db.delete(access_token)
+    with pytest.raises(ValueError, match="Access token could not be deleted"):
+        await dynamodb_access_token_db.delete(access_token)
+
     deleted_token = await dynamodb_access_token_db.get_by_token(access_token.token)
     assert deleted_token is None
 

tests/test_users.py

@@ -107,6 +107,8 @@ async def test_queries(dynamodb_user_db: DynamoDBUserDatabase[User, UUID_ID]):
 
     # Delete user
     await dynamodb_user_db.delete(user)
+    with pytest.raises(ValueError, match="User account could not be deleted"):
+        await dynamodb_user_db.delete(user)
     deleted_user = await dynamodb_user_db.get(user.id)
     assert deleted_user is None
 
@@ -160,6 +162,7 @@ async def test_queries_oauth(
     dynamodb_user_db_oauth: DynamoDBUserDatabase[UserOAuth, UUID_ID],
     oauth_account1: dict[str, Any],
     oauth_account2: dict[str, Any],
+    user_id: UUID_ID,
 ):
     # Test OAuth accounts
     user_create = {"email": "lancelot@camelot.bt", "hashed_password": "guinevere"}
@@ -216,3 +219,19 @@ def _get_account(_user: UserOAuth):
     # Unknown OAuth account
     unknown_oauth_user = await dynamodb_user_db_oauth.get_by_oauth_account("foo", "bar")
     assert unknown_oauth_user is None
+
+    with pytest.raises(
+        ValueError,
+        match="OAuth account could not be updated because it does not exist.",
+    ):
+        user = UserOAuth()
+        oauth_account = OAuthAccount()
+        oauth_account.user_id = user_id
+        oauth_account.oauth_name = "blabla_provider"
+        oauth_account.account_id = "blabla_id"
+        oauth_account.account_email = "blabla@gmail.com"
+        await dynamodb_user_db_oauth.update_oauth_account(
+            user,
+            oauth_account,
+            {"access_token": "NEW_TOKEN"},
+        )