better tests and create conditions

Author: Jade Wibbels

app/routes/posts.py

@@ -3,6 +3,8 @@
 import boto3
 from botocore.exceptions import ClientError
 
+from app.routes.users import is_existing_user_id
+
 router = APIRouter()
 
 # Initialize the DynamoDB client
@@ -17,6 +19,14 @@ class Post(BaseModel):
     author_id: str
 
 
+def is_existing_post_id(post_Id: str):
+    try:
+        response = table.scan(FilterExpression="Id = :i", ExpressionAttributeValues={":i": post_Id})
+        return len(response["Items"]) == 1
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
 @router.get("/")
 def read_posts():
     """
@@ -72,6 +82,10 @@ def create_post(post: Post):
     ``{"message": "Post created"}``
     """
     try:
+        if not is_existing_user_id(post.author_id):
+            raise HTTPException(status_code=400, detail="Cannot create post for unknown User Id.")
+        if is_existing_post_id(post.Id):
+            raise HTTPException(status_code=400, detail="Cannot create post for duplicate Id.")
         table.put_item(Item={k: v for k, v in post.model_dump().items()})
 
         return {"message": "Post created"}
@@ -97,6 +111,8 @@ def update_post(Id: str, post: Post):
     ``{"message": "Post with id {Id} updated", "updated_attributes": {updated_attributes}}``
     """
     try:
+        if not is_existing_post_id(post.Id):
+            raise HTTPException(status_code=400, detail="Cannot update post for unknown Id.")
         response = table.update_item(
             Key={"Id": Id},
             UpdateExpression="set title=:t, content=:c, author_id=:a",
@@ -123,6 +139,8 @@ def delete_post(Id: str):
     ``{"message":"Post with id 1 deleted"}
     """
     try:
+        if not is_existing_post_id(Id):
+            raise HTTPException(status_code=400, detail="Cannot delete post for unknown Id.")
         table.delete_item(Key={"Id": Id})
         return {"message": f"Post with id {Id} deleted"}
     except ClientError as e:

app/routes/users.py

@@ -17,6 +17,36 @@ class User(BaseModel):
     email_address: str
 
 
+def is_existing_user_id(user_Id: str):
+    try:
+        response = table.scan(FilterExpression="Id = :i", ExpressionAttributeValues={":i": user_Id})
+        return len(response["Items"]) == 1
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+def is_existing_user_name(user_name: str):
+    try:
+        response = table.scan(
+            FilterExpression="user_name = :u",
+            ExpressionAttributeValues={":u": user_name},
+        )
+        return len(response["Items"]) > 0
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+def is_existing_email_address(email_address: str):
+    try:
+        response = table.scan(
+            FilterExpression="email_address = :e",
+            ExpressionAttributeValues={":e": email_address},
+        )
+        return len(response["Items"]) > 0
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
 @router.post("/")
 def create_user(user: User):
     """
@@ -33,6 +63,12 @@ def create_user(user: User):
     Response: ``{"message":"User created"}``
     """
     try:
+        # Check if user_name or email_address already exists
+        if is_existing_user_name(user.user_name):
+            raise HTTPException(status_code=400, detail="User name already exists")
+        if is_existing_email_address(user.email_address):
+            raise HTTPException(status_code=400, detail="Email address already exists")
+
         table.put_item(
             Item={
                 "Id": user.Id,
@@ -47,7 +83,7 @@ def create_user(user: User):
 
 
 @router.get("/{Id}")
-def read_user(Id: str):
+def read_user_by_id(Id: str):
     """
     Reads a user from the DynamoDB table "users" by Id
 
@@ -98,6 +134,8 @@ def update_user(Id: str, user: User):
     Response: ``{"message":"User with id 1 updated","updated_attributes": ...}``
     """
     try:
+        if not is_existing_user_id(Id):
+            raise HTTPException(status_code=400, detail="User Id not found.")
         response = table.update_item(
             Key={"Id": Id},
             UpdateExpression="set user_name=:u, password=:p, email_address=:e",
@@ -126,6 +164,8 @@ def delete_user(Id: str):
     Response: ``{"message":"User with id 1 deleted"}``
     """
     try:
+        if not is_existing_user_id(Id):
+            raise HTTPException(status_code=400, detail="User Id not found.")
         table.delete_item(Key={"Id": Id})
         return {"message": f"User with id {Id} deleted"}
     except ClientError as e:

requirements.txt

@@ -1,3 +1,4 @@
+moto == 5.0.22
 pytest
 pytest-asyncio
 pytest-cov

tests/conftest.py

@@ -1,19 +1,30 @@
 import pytest
-from botocore.stub import Stubber
 from fastapi.testclient import TestClient
+from moto import mock_aws
+import boto3
 
 from app.main import app
 
 
 @pytest.fixture
-def client():
+def client(ddb_client):
+    dynamodb = get_dynamodb(ddb_client)  # noqa F841
     with TestClient(app) as client:
         yield client
 
 
-@pytest.fixture
-def mock_dynamodb(request):
-    table = request.param
-    with Stubber(table.meta.client) as stubber:
-        yield stubber
-        stubber.assert_no_pending_responses()
+@pytest.fixture(scope="function")
+def ddb_client():
+    with mock_aws():
+        yield boto3.client("dynamodb")
+
+
+def get_dynamodb(ddb_client):
+    tables = ["users", "posts"]
+    for table_name in tables:
+        ddb_client.create_table(
+            TableName=table_name,
+            KeySchema=[{"AttributeName": "Id", "KeyType": "HASH"}],
+            AttributeDefinitions=[{"AttributeName": "Id", "AttributeType": "S"}],
+            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+        )