diff --git a/python-test-samples/README.md b/python-test-samples/README.md index e25b76f2..04344bfc 100644 --- a/python-test-samples/README.md +++ b/python-test-samples/README.md @@ -12,6 +12,7 @@ This portion of the repository contains code samples for testing serverless appl |[Lambda local testing with Mocks](./lambda-mock)|This project contains unit tests for Lambda using mocks.| |[Lambda Layers with Mocks](./apigw-lambda-layer)|This project contains unit tests for Lambda layers using mocks.| |[API Gateway with Lambda and DynamoDB](./apigw-lambda-dynamodb)|This project contains unit and integration tests for a pattern using API Gateway, AWS Lambda and Amazon DynamoDB.| +|[API Gateway with local Lambda and local DynamoDB](./apigw-lambda-dynamodb-crud-local)|This project contains unit test for local execution CRUD paterns using API Gateway, AWS Lambda and Amazon DynamoDB.| |[Schema and Contract Testing](./schema-and-contract-testing)|This project contains sample schema and contract tests for an event driven architecture.| |[Kinesis with Lambda and DynamoDB](./kinesis-lambda-dynamodb)|This project contains a example of testing an application with an Amazon Kinesis Data Stream.| |[SQS with Lambda](./apigw-sqs-lambda-sqs)|This project demonstrates testing SQS as a source and destination in an integration test| diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/README.md b/python-test-samples/apigw-lambda-dynamodb-crud-local/README.md new file mode 100644 index 00000000..ec8b30a0 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/README.md @@ -0,0 +1,408 @@ +[![python: 3.10](https://img.shields.io/badge/Python-3.10-green)](https://img.shields.io/badge/Python-3.10-green) +[![AWS: DynamoDB](https://img.shields.io/badge/AWS-DynamoDB-blueviolet)](https://img.shields.io/badge/AWS-DynamoDB-blueviolet) +[![AWS: Lambda](https://img.shields.io/badge/AWS-Lambda-orange)](https://img.shields.io/badge/AWS-Lambda-orange) +[![AWS: API Gateway](https://img.shields.io/badge/AWS-API%20Gateway-blue)](https://img.shields.io/badge/AWS-API%20Gateway-blue) +[![test: pytest](https://img.shields.io/badge/Test-Pytest-red)](https://img.shields.io/badge/Test-Pytest-red) +[![test: local](https://img.shields.io/badge/Test-Local-red)](https://img.shields.io/badge/Test-Local-red) + +# Local: AWS API Gateway, Lambda, and DynamoDB Integration Testing + +## Introduction + +This project demonstrates how to test AWS serverless applications locally on Docker using PyTest. It implements comprehensive testing for a CRUD API built with API Gateway, Lambda, and DynamoDB, showcasing integration testing patterns and local service emulation, all tested through SAM Emulation and PyTest. + +--- + +## Contents +- [Local: AWS API Gateway, Lambda, and DynamoDB Integration Testing](#local-aws-api-gateway-lambda-and-dynamodb-integration-testing) + - [Introduction](#introduction) + - [Contents](#contents) + - [Architecture Overview](#architecture-overview) + - [Project Structure](#project-structure) + - [Prerequisites](#prerequisites) + - [Test Scenarios](#test-scenarios) + - [About the Test Process](#about-the-test-process) + - [API Endpoints](#api-endpoints) + - [Testing Workflows](#testing-workflows) + - [Common Issues](#common-issues) + - [Additional Resources](#additional-resources) + +--- + +## Architecture Overview +

+ API Gateway CRUD Architecture +

+ +Components: +- API Gateway endpoints for CRUD operations +- Lambda functions for business logic +- DynamoDB table for data persistence +- PyTest framework for automated testing +- Docker containers for local service emulation + +--- + +## Project Structure +``` +├── events _# folder containing json files for API Gateway CRUD input events_ +├── img/apigateway-crud-lambda-dynamodb.png _# Architecture diagram_ +├── lambda_crud_src _# folder containing code for different CRUD Lambda functions_ +├── tests/ +│ ├── unit/src/test_crud_operations.py _# python PyTest test definition_ +│ └── requirements.txt _# pip requirements dependencies file_ +├── template.yaml _# sam yaml template file for necessary components test_ +└── README.md _# instructions file_ +``` + +--- + +## Prerequisites +- Docker +- Python 3.9 or newer (running pytest) +- AWS SAM CLI (running SAM Lambda emulator) +- curl (for debugging) +- Basic understanding of API Gateway, Lambda Functions and DynamoDB + +--- + +## Test Scenarios + +### 1. CRUD Operations +- Tests complete Create, Read, Update, Delete cycle +- Validates response structures and status codes +- Verifies data persistence in DynamoDB + +### 2. Error Handling +- Tests invalid inputs +- Verifies error responses +- Validates error handling middleware + +### 3. Integration Flow +- Tests end-to-end request processing +- Validates service integration points +- Verifies transaction consistency + +--- + +## About the Test Process + +The test process leverages PyTest fixtures to manage service lifecycles: + +1. **Service Setup**: + - Launches DynamoDB Local container + - Starts SAM Local API Gateway Emulator + - Initializes test database + +2. **Test Execution**: + - Runs CRUD operation tests + - Validates responses and data + - Verifies error scenarios + +3. **Cleanup**: + - Removes test data + - Stops containers + - Cleans up resources + +--- + +## API Endpoints + +| Endpoint | Method | Description | +|----------|--------|-------------| +| /init | GET | Creates DynamoDB CRUDLocalTable table | +| /create | POST | Creates new item | +| /read | GET | Retrieves an item | +| /update | POST | Updates existing item | +| /delete | GET | Deletes an item | + +--- + +## Testing Workflows + +### Setup Docker Environment + +> Make sure docker engine is running before running the tests. + +``` shell +apigw-lambda-dynamodb-crud-local$ docker version +Client: Docker Engine - Community + Version: 24.0.6 + API version: 1.43 +``` + +### Run the Unit Test - End to end python test + +> Configure environment variables: + +``` shell +apigw-lambda-dynamodb-crud-local$ +export AWS_ACCESS_KEY_ID='DUMMYIDEXAMPLE' +export AWS_SECRET_ACCESS_KEY='DUMMYEXAMPLEKEY' +export AWS_REGION='us-east-1' +``` + +> Start the DynamoDB Container and SAM Local Lambda emulator in a separate terminal: + +```shell +# Start DynamoDB Local +docker run --rm -d -p 8000:8000 --name dynamodb-local amazon/dynamodb-local + +# Start SAM Local API Gateway emulator: +sam local start-api --docker-network host & +``` + +> Set up the python environment: + +``` shell +apigw-lambda-dynamodb-crud-local$ +python3 -m venv venv +source venv/bin/activate +pip install --upgrade pip +pip install -r tests/requirements.txt +``` + +#### Run the Unit Tests + +``` shell +apigw-lambda-dynamodb-crud-local$ +python3 -m pytest -s tests/unit/src/test_crud_operations.py +``` + +Expected output +``` shell +apigw-lambda-dynamodb-crud-local$ +python3 -m pytest -s tests/unit/src/test_crud_operations.py +================================================================= test session starts ================================================================== +platform linux -- Python 3.10.12, pytest-7.4.4, pluggy-1.6.0 +rootdir: /home/ubuntu/environment/python-test-samples/apigw-lambda-dynamodb-crud-local +plugins: timeout-2.1.0 +collected 10 items + +tests/unit/src/test_crud_operations.py DynamoDB Local is already running +SAM Local API Gateway is running and responding +=== Step 1: Initialize DynamoDB Table === +✓ Table already exists (expected): 500 +Response: Error creating table: An error occurred (ResourceInUseException) when calling the CreateTable operation: Cannot create preexisting table + +.=== Step 2: Create Initial Item === +✓ Item creation successful: 200 +Created item: {'Id': '123', 'name': 'Batman'} +Response: {"message": "Item added", "response": {"ResponseMetadata": {"RequestId": "eaf2a56a-0c8e-4d3a-833a-1e33dd171fc2", "HTTPStatusCode": 200, "HTTPHeaders": {"server": "Jetty(12.0.14)", "date": "Mon, 04 Aug 2025 13:11:58 GMT", "x-amzn-requestid": "eaf2a56a-0c8e-4d3a-833a-1e33dd171fc2", "content-type": "application/x-amz-json-1.0", "x-amz-crc32": "2745614147", "content-length": "2"}, "RetryAttempts": 0}}} + +.=== Step 3: Read Item === +✓ Item read successful: Id=123, name=Batman +Response: {"name": "Batman", "Id": "123"} + +.=== Step 4: Update Item === +✓ Item update successful: 200 +Updated to: {'Id': '123', 'name': 'Robin'} +Response: {"message": "Item updated successfully", "response": {"Attributes": {"name": "Robin"}, "ResponseMetadata": {"RequestId": "6bd871d4-c87d-4b16-a63d-20d08051cae7", "HTTPStatusCode": 200, "HTTPHeaders": {"server": "Jetty(12.0.14)", "date": "Mon, 04 Aug 2025 13:12:00 GMT", "x-amzn-requestid": "6bd871d4-c87d-4b16-a63d-20d08051cae7", "content-type": "application/x-amz-json-1.0", "x-amz-crc32": "945407983", "content-length": "37"}, "RetryAttempts": 0}}} + +.=== Step 5: Check Updated Item === +✓ Updated item read successful: name=Robin +Response: {"name": "Robin", "Id": "123"} + +.=== Step 6: Delete Item === +✓ Item deletion successful: 200 +Deleted item: {'Id': '123'} +Response: {"message": "Item deleted", "response": {"ResponseMetadata": {"RequestId": "b59cad53-1185-4b3c-9a69-388b731aeb5a", "HTTPStatusCode": 200, "HTTPHeaders": {"server": "Jetty(12.0.14)", "date": "Mon, 04 Aug 2025 13:12:02 GMT", "x-amzn-requestid": "b59cad53-1185-4b3c-9a69-388b731aeb5a", "content-type": "application/x-amz-json-1.0", "x-amz-crc32": "2745614147", "content-length": "2"}, "RetryAttempts": 0}}} + +.=== Step 7: Verify Item Deleted === +✓ Item correctly deleted: 404 Not Found +Response: {"error": "Item not found", "message": "No item with Id 123 found"} + +=== CRUD Sequence Complete! === +.=== Complete Integration Cycle Test === +Testing with ID: integration-456 +Step 1: Create item +✓ Create: 200 +Step 2: Read item +✓ Read: 200 +Step 3: Update item +✓ Update: 200 +Step 4: Verify update +✓ Verify: 200 +Step 5: Delete item +✓ Delete: 200 +Step 6: Verify deletion +✓ Verify deletion: 404 +✓ Complete integration cycle passed! + +.=== Error Scenarios Test === +Test 1: Read non-existent item +✓ Read nonexistent: 404 +Test 2: Update non-existent item +✓ Update nonexistent: 200 +Test 3: Delete non-existent item +✓ Delete nonexistent: 200 +✓ Error scenarios completed + +.=== Performance Check === +✓ Create: 718ms (status: 200) +✓ Read: 724ms (status: 200) +✓ Update: 740ms (status: 200) +✓ Delete: 729ms (status: 200) +✓ Average operation time: 728ms +✓ Performance check completed + +. + +================================================================= 10 passed in 36.91s ================================================================== +``` + +#### Clean up section + +> clean pyenv environment + +```sh +apigw-lambda-dynamodb-crud-local$ +deactivate +rm -rf venv/ +``` + +> unsetting variables + +```sh +unset AWS_ACCESS_KEY_ID +unset AWS_SECRET_ACCESS_KEY +unset AWS_REGION +``` + +> cleaning sam process + +```sh +ps -axuf | grep '[s]am local start-api' | awk '{print $2}' | xargs -r kill +``` + +> cleanning docker + +```sh +docker ps -q --filter ancestor=amazon/dynamodb-local | xargs -r docker kill +docker rmi amazon/dynamodb-local +``` + +--- + +#### Debug - PyTest Debugging + +For more detailed debugging in pytest: + +```sh +# Run with verbose output +python3 -m pytest -s -v unit/src/test_crud_operations.py + +# Run with debug logging +python3 -m pytest -s tests/unit/src/test_crud_operations.py --log-cli-level=DEBUG + +# List available individual test +python3 -m pytest tests/unit/src/test_crud_operations.py --collect-only + +# Run a specific pytest test +python3 -m pytest -s tests/unit/src/test_crud_operations.py::test_01_initialize_table -v + +``` + +### Fast local development for CRUD Operations + +#### AWS CLI Commands for Manual Verification + +If you need to manually verify the CRUD Operations or execution details, you can use these commands: + +#### Configure environment variables: + +``` shell +apigw-lambda-dynamodb-crud-local$ +export AWS_ACCESS_KEY_ID='DUMMYIDEXAMPLE' +export AWS_SECRET_ACCESS_KEY='DUMMYEXAMPLEKEY' +export AWS_REGION='us-east-1' +``` + +#### Start the DynamoDB Container and SAM Local Lambda emulator in a separate terminal: + +```shell +# Start DynamoDB Local +docker run --rm -d -p 8000:8000 --name dynamodb-local amazon/dynamodb-local + +# Start SAM Local API Gateway emulator: +sam local start-api --docker-network host & +``` + +#### Debug lambda functions - Test Individually API and Lambda Functions + +0. Initialize the DynamoDB table (though Api Gateway -> Lambda crud init function): +```sh +curl -X GET http://127.0.0.1:3000/init +``` + +1. Create initial item: +```sh +curl -X POST http://127.0.0.1:3000/create \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123", "name": "Batman"}' +``` + +2. Read item: +```sh +curl -X GET http://127.0.0.1:3000/read \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' +``` + +3. Update initial item: +```sh +curl -X POST http://127.0.0.1:3000/update \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123", "name": "Robin"}' +``` + +4. Check updated item: +```sh +curl -X GET http://127.0.0.1:3000/read \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' +``` + +5. Delete item: +```sh +curl -X GET http://127.0.0.1:3000/delete \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' +``` + +6. Checking item does not exist: +```sh +curl -X GET http://127.0.0.1:3000/read \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' +``` + +--- + +## Common Issues + +### DynamoDB Issues +- Verify DynamoDB Local Container is running +- Check port 8000 availability +- Confirm network settings (no using host network) +- Table exist initalization error (Init request failed with status 500: Error creating table: An error occurred (ResourceInUseException) when calling the CreateTable operation: Cannot create preexisting table) -> Clean up the dynamodb docker image and related columnes + +### SAM Local API Issues +- Ensure template.yaml is valid +- Verify Lambda function handlers +- Check Docker network configuration (using host network) + +### PyTest Failures +- Verify Python environment (recreation) +- Check fixture dependencies (requirements) +- Review test isolation + +--- + +## Additional Resources +- [PyTest Documentation](https://docs.pytest.org/) +- [AWS SAM Local Testing](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-testing.html) +- [DynamoDB Local Guide](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBLocal.html) +- [API Gateway Testing](https://docs.aws.amazon.com/apigateway/latest/developerguide/how-to-test-api.html) + +[Top](#contents) diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-create-event.json b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-create-event.json new file mode 100755 index 00000000..1c47303f --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-create-event.json @@ -0,0 +1,3 @@ +{ + "body": "{\"Id\": \"123\", \"name\": \"Batman\"}" +} diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-delete-event.json b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-delete-event.json new file mode 100755 index 00000000..12090c66 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-delete-event.json @@ -0,0 +1,3 @@ +{ + "Id": "123" +} diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-init-event.json b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-init-event.json new file mode 100755 index 00000000..4ccb6ff4 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-init-event.json @@ -0,0 +1,5 @@ +{ + "key1": "value1", + "key2": "value2", + "key3": "value3" +} \ No newline at end of file diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-read-event.json b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-read-event.json new file mode 100755 index 00000000..12090c66 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-read-event.json @@ -0,0 +1,3 @@ +{ + "Id": "123" +} diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-update-event.json b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-update-event.json new file mode 100755 index 00000000..074569d3 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/events/lambda-update-event.json @@ -0,0 +1,3 @@ +{ + "body": "{\"Id\": \"123\", \"name\": \"Robin\"}" +} diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/img/apigateway-crud-lambda-dynamodb.png b/python-test-samples/apigw-lambda-dynamodb-crud-local/img/apigateway-crud-lambda-dynamodb.png new file mode 100644 index 00000000..91cd84de Binary files /dev/null and b/python-test-samples/apigw-lambda-dynamodb-crud-local/img/apigateway-crud-lambda-dynamodb.png differ diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_create_src/app.py b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_create_src/app.py new file mode 100755 index 00000000..8a586f08 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_create_src/app.py @@ -0,0 +1,22 @@ +import os +import json +import boto3 +def lambda_handler(event, context): + # Checking if running locally + if os.environ.get('AWS_SAM_LOCAL'): + # Use local DynamoDB endpoint + dynamodb = boto3.resource('dynamodb', endpoint_url='http://172.17.0.1:8000') + else: + # Use the default DynamoDB endpoint (AWS) + dynamodb = boto3.resource('dynamodb') + + # Adding an item to DynamoDB + item = json.loads(event['body']) + + table = dynamodb.Table(os.environ['DYNAMODB_TABLE']) + response = table.put_item(Item=item) + + return { + 'statusCode': 200, + 'body': json.dumps({'message': 'Item added', 'response': response}) + } diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_delete_src/app.py b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_delete_src/app.py new file mode 100755 index 00000000..22ff01ef --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_delete_src/app.py @@ -0,0 +1,27 @@ +import os +import json +import boto3 + +def lambda_handler(event, context): + # Checking if running locally + if os.environ.get('AWS_SAM_LOCAL'): + # Use local DynamoDB endpoint + dynamodb = boto3.resource('dynamodb', endpoint_url='http://172.17.0.1:8000') + else: + # Use the default DynamoDB endpoint (AWS) + dynamodb = boto3.resource('dynamodb') + + # Parse the JSON body + body = json.loads(event['body']) + + # Access the Id element + item_id = body['Id'] + + # Deleting item on DynamoDB + table = dynamodb.Table(os.environ['DYNAMODB_TABLE']) + response = table.delete_item(Key={'Id': item_id}) + + return { + 'statusCode': 200, + 'body': json.dumps({'message': 'Item deleted', 'response': response}) + } diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_init_src/app.py b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_init_src/app.py new file mode 100755 index 00000000..29ef4fa4 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_init_src/app.py @@ -0,0 +1,52 @@ +import os +import boto3 + +def lambda_handler(event, context): + # Check if running locally + if os.environ.get('AWS_SAM_LOCAL'): + # Use local DynamoDB endpoint + dynamodb = boto3.resource('dynamodb', endpoint_url='http://172.17.0.1:8000') + else: + # Use the default DynamoDB endpoint (AWS) + dynamodb = boto3.resource('dynamodb') + + # Access your DynamoDB table + table_name = dynamodb.Table(os.environ['DYNAMODB_TABLE']) + + # Define the table creation parameters + table_creation_params = { + 'TableName': 'CRUDLocalTable', + 'KeySchema': [ + { + 'AttributeName': 'Id', + 'KeyType': 'HASH' # Partition key + } + ], + 'AttributeDefinitions': [ + { + 'AttributeName': 'Id', + 'AttributeType': 'S' # String type + } + ], + 'ProvisionedThroughput': { + 'ReadCapacityUnits': 5, + 'WriteCapacityUnits': 5 + } + } + + try: + # Create the DynamoDB table + table = dynamodb.create_table(**table_creation_params) + # Wait until the table exists before continuing + table.wait_until_exists() + print("Table created successfully:", table.table_name) + return { + 'statusCode': 200, + 'body': table.table_name + } + except Exception as e: + print(f"Error creating table: {e}") + return { + 'statusCode': 500, + 'body': f"Error creating table: {e}" + } diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_read_src/app.py b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_read_src/app.py new file mode 100644 index 00000000..f9045df8 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_read_src/app.py @@ -0,0 +1,46 @@ +import os +import boto3 +import json + +def lambda_handler(event, context): + # Check if running locally + if os.environ.get('AWS_SAM_LOCAL'): + # Use local DynamoDB endpoint + dynamodb = boto3.resource('dynamodb', endpoint_url='http://172.17.0.1:8000') + else: + # Use the default DynamoDB endpoint (AWS) + dynamodb = boto3.resource('dynamodb') + + # Access your DynamoDB table + table = dynamodb.Table(os.environ['DYNAMODB_TABLE']) + + # Parse the JSON body + body = json.loads(event['body']) + + # Access the Id element + item_id = body['Id'] + + try: + response = table.get_item(Key={'Id': item_id}) + + item = response.get('Item', {}) + + if item: + # Item found, return it + return { + 'statusCode': 200, + 'body': json.dumps(item) + } + else: + # Item not found, return a 404 status + return { + 'statusCode': 404, + 'body': json.dumps({'error': 'Item not found', 'message': f'No item with Id {item_id} found'}) + } + + except Exception as e: + # Failed reading from DynamoDB + return { + 'statusCode': 500, + 'body': json.dumps({'error': 'Failed to read item', 'message': str(e)}) + } diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_update_src/app.py b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_update_src/app.py new file mode 100755 index 00000000..3a6900a8 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/lambda_crud_src/lambda_crud_update_src/app.py @@ -0,0 +1,55 @@ +import os +import json +import boto3 + +def lambda_handler(event, context): + # Check if running locally + if os.environ.get('AWS_SAM_LOCAL'): + # Use local DynamoDB endpoint + dynamodb = boto3.resource('dynamodb', endpoint_url='http://172.17.0.1:8000') + else: + # Use the default DynamoDB endpoint (AWS) + dynamodb = boto3.resource('dynamodb') + + + # Parse the incoming event + try: + body = json.loads(event['body']) # Parse the JSON string in 'body' + item_id = body['Id'] # Access the 'Id' from the parsed body + item_name = body['name'] # Access the 'name' from the parsed body + except (json.JSONDecodeError, KeyError) as e: + return { + 'statusCode': 400, + 'body': json.dumps({'error': 'Invalid input', 'message': str(e)}) + } + + # Reference the DynamoDB table + table = dynamodb.Table(os.environ['DYNAMODB_TABLE']) + + # Update item in DynamoDB + try: + response = table.update_item( + Key={'Id': item_id}, + UpdateExpression="SET #name = :name", + ExpressionAttributeNames={ + '#name': 'name' # Using an alias for reserved words + }, + ExpressionAttributeValues={ + ':name': item_name + }, + ReturnValues="UPDATED_NEW" + ) + + return { + 'statusCode': 200, + 'body': json.dumps({ + 'message': 'Item updated successfully', + 'response': response + }) + } + + except Exception as e: + return { + 'statusCode': 500, + 'body': json.dumps({'error': 'Failed to update item', 'message': str(e)}) + } diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/template.yaml b/python-test-samples/apigw-lambda-dynamodb-crud-local/template.yaml new file mode 100755 index 00000000..8fcd1cca --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/template.yaml @@ -0,0 +1,131 @@ +AWSTemplateFormatVersion: '2010-09-09' +Transform: AWS::Serverless-2016-10-31 +Description: AWS SAM Template for CRUD operations with API Gateway and Python Lambda Functions and Local DynamoDB + +Resources: + # API Gateway with CRUD integration + APIGatewayCRUD: + Type: AWS::Serverless::Api + Properties: + StageName: Prod + + # DynamoDB table persisting CRUD operations + CRUDLocalTable: + Type: AWS::DynamoDB::Table + Properties: + TableName: CRUDLocalTable + AttributeDefinitions: + - AttributeName: Id + AttributeType: S + KeySchema: + - AttributeName: Id + KeyType: HASH + ProvisionedThroughput: + ReadCapacityUnits: 5 + WriteCapacityUnits: 5 + + # Lambda function to initialize the DynamoDB table + CRUDLambdaInitFunction: + Type: AWS::Serverless::Function + Properties: + CodeUri: lambda_crud_src/lambda_crud_init_src + Handler: app.lambda_handler + Runtime: python3.9 + Timeout: 30 + Environment: + Variables: + DYNAMODB_TABLE: CRUDLocalTable + AWS_ACCESS_KEY_ID: DUMMYIDEXAMPLE + AWS_SECRET_ACCESS_KEY: DUMMYEXAMPLEKEY + REGION: eu-west-1 + Events: + EventInit: + Type: Api + Properties: + Path: /init + Method: get + RestApiId: !Ref APIGatewayCRUD + + # Lambda function creating items on DynamoDB table + CRUDLambdaCreateFunction: + Type: AWS::Serverless::Function + Properties: + CodeUri: lambda_crud_src/lambda_crud_create_src + Handler: app.lambda_handler + Runtime: python3.9 + Environment: + Variables: + DYNAMODB_TABLE: CRUDLocalTable + AWS_ACCESS_KEY_ID: DUMMYIDEXAMPLE + AWS_SECRET_ACCESS_KEY: DUMMYEXAMPLEKEY + REGION: eu-west-1 + Events: + EventCreate: + Type: Api + Properties: + Path: /create + Method: post + RestApiId: !Ref APIGatewayCRUD + + # Lambda function reading items on DynamoDB table + CRUDLambdaReadFunction: + Type: AWS::Serverless::Function + Properties: + CodeUri: lambda_crud_src/lambda_crud_read_src + Handler: app.lambda_handler + Runtime: python3.9 + Environment: + Variables: + DYNAMODB_TABLE: CRUDLocalTable + AWS_ACCESS_KEY_ID: DUMMYIDEXAMPLE + AWS_SECRET_ACCESS_KEY: DUMMYEXAMPLEKEY + REGION: eu-west-1 + Events: + EventGet: + Type: Api + Properties: + Path: /read + Method: get + RestApiId: !Ref APIGatewayCRUD + + # Lambda function deleting items on DynamoDB table + CRUDLambdaDeleteFunction: + Type: AWS::Serverless::Function + Properties: + CodeUri: lambda_crud_src/lambda_crud_delete_src + Handler: app.lambda_handler + Runtime: python3.9 + Environment: + Variables: + DYNAMODB_TABLE: CRUDLocalTable + AWS_ACCESS_KEY_ID: DUMMYIDEXAMPLE + AWS_SECRET_ACCESS_KEY: DUMMYEXAMPLEKEY + REGION: eu-west-1 + Events: + EventDelete: + Type: Api + Properties: + Path: /delete + Method: get + RestApiId: !Ref APIGatewayCRUD + + # Lambda function updating items on DynamoDB table + CRUDLambdaUpdateFunction: + Type: AWS::Serverless::Function + Properties: + CodeUri: lambda_crud_src/lambda_crud_update_src + Handler: app.lambda_handler + Runtime: python3.9 + Environment: + Variables: + DYNAMODB_TABLE: CRUDLocalTable + AWS_ACCESS_KEY_ID: DUMMYIDEXAMPLE + AWS_SECRET_ACCESS_KEY: DUMMYEXAMPLEKEY + REGION: eu-west-1 + Events: + EventUpdate: + Type: Api + Properties: + Path: /update + Method: post + RestApiId: !Ref APIGatewayCRUD diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/tests/requirements.txt b/python-test-samples/apigw-lambda-dynamodb-crud-local/tests/requirements.txt new file mode 100644 index 00000000..c8e81ba5 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/tests/requirements.txt @@ -0,0 +1,20 @@ +# Core testing framework +pytest==7.4.4 + +# HTTP client for API requests (replaces curl commands) +requests==2.31.0 + +# Docker container management for DynamoDB Local +docker==7.0.0 + +# Basic testing utilities +pytest-timeout==2.1.0 + +# JSON processing and validation +jsonschema==4.21.1 + +# Date/time utilities (used in test) +python-dateutil==2.8.2 + +# Better output formatting (optional but helpful) +colorlog==6.8.2 \ No newline at end of file diff --git a/python-test-samples/apigw-lambda-dynamodb-crud-local/tests/unit/src/test_crud_operations.py b/python-test-samples/apigw-lambda-dynamodb-crud-local/tests/unit/src/test_crud_operations.py new file mode 100644 index 00000000..d0c10721 --- /dev/null +++ b/python-test-samples/apigw-lambda-dynamodb-crud-local/tests/unit/src/test_crud_operations.py @@ -0,0 +1,596 @@ +import pytest +import requests +import json +import time +import docker +import socket +from datetime import datetime + + +@pytest.fixture(scope="session") +def dynamodb_local(): + """ + Fixture to start DynamoDB Local container. + Reproduces: docker run --rm -d --network host -p 8000:8000 amazon/dynamodb-local + """ + client = docker.from_env() + + # Check if DynamoDB Local is already running + try: + response = requests.get("http://localhost:8000", timeout=5) + print("DynamoDB Local is already running") + yield "http://localhost:8000" + return + except requests.exceptions.RequestException: + pass + + # Start DynamoDB Local container with port mapping (fixed networking) + try: + container = client.containers.run( + "amazon/dynamodb-local", + ports={'8000/tcp': 8000}, # Fixed: proper port mapping without host networking + detach=True, + remove=True, + name=f"dynamodb-local-api-test-{int(time.time())}" + ) + + # Wait for DynamoDB to be ready + max_retries = 30 + for i in range(max_retries): + try: + response = requests.get("http://localhost:8000", timeout=2) + if response.status_code == 400: # DynamoDB returns 400 for root path + print("DynamoDB Local container is ready") + break + except requests.exceptions.RequestException: + time.sleep(1) + if i == max_retries - 1: + container.stop() + pytest.fail("DynamoDB Local container failed to start") + + yield "http://localhost:8000" + + # Cleanup + try: + container.stop() + container.remove() + except: + pass + + except docker.errors.ImageNotFound: + pytest.skip("DynamoDB Local Docker image not available") + except Exception as e: + pytest.skip(f"Failed to start DynamoDB Local container: {str(e)}") + + +@pytest.fixture(scope="session") +def sam_local_api(dynamodb_local): + """ + Fixture to verify SAM Local API Gateway is running. + Assumes: sam local start-api --docker-network host & + """ + # Check if SAM Local API Gateway is running on port 3000 + def is_port_open(host, port): + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(5) + result = s.connect_ex((host, port)) + return result == 0 + except: + return False + + max_retries = 10 + for i in range(max_retries): + if is_port_open("127.0.0.1", 3000): + try: + # Try to make a test request to verify API Gateway is responding + response = requests.get("http://127.0.0.1:3000/init", timeout=10) + print("SAM Local API Gateway is running and responding") + break + except requests.exceptions.RequestException: + if i == max_retries - 1: + pytest.skip("SAM Local API Gateway is not responding. Please start with 'sam local start-api --docker-network host'") + else: + if i == max_retries - 1: + pytest.skip("SAM Local API Gateway is not running on port 3000. Please start with 'sam local start-api --docker-network host'") + time.sleep(2) + + yield "http://127.0.0.1:3000" + + +@pytest.fixture(scope="session") +def api_client(): + """ + Fixture to create HTTP client for API requests. + """ + session = requests.Session() + session.headers.update({ + 'Content-Type': 'application/json', + 'User-Agent': 'pytest-api-integration-test' + }) + return session + + +def test_01_initialize_table(sam_local_api, api_client, dynamodb_local): + """ + Test initializing the DynamoDB table (idempotent). + Reproduces: curl -X GET http://127.0.0.1:3000/init + + Based on Lambda init code: + - Returns 200 with table name when successfully created + - Returns 500 with error message when table already exists or other error + """ + base_url = sam_local_api + + print("=== Step 1: Initialize DynamoDB Table ===") + + # Initialize the DynamoDB table via API Gateway + response = api_client.get(f"{base_url}/init") + + # Handle both success and "already exists" scenarios (idempotent) + if response.status_code == 200: + # Table created successfully + print(f"✓ Table created successfully: {response.status_code}") + assert 'CRUDLocalTable' in response.text or response.text.strip() == 'CRUDLocalTable', \ + f"Response should contain table name: {response.text}" + + elif response.status_code == 500: + # Check if it's the expected "table already exists" error + error_text = response.text.lower() + if any(phrase in error_text for phrase in ['already exists', 'preexisting', 'resourceinuse', 'cannot create']): + print(f"✓ Table already exists (expected): {response.status_code}") + else: + pytest.fail(f"Unexpected 500 error: {response.text}") + + else: + pytest.fail(f"Unexpected response status {response.status_code}: {response.text}") + + print(f"Response: {response.text}") + print() + + +def test_02_create_item(sam_local_api, api_client): + """ + Test creating an item. + Reproduces: curl -X POST http://127.0.0.1:3000/create \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123", "name": "Batman"}' + """ + base_url = sam_local_api + + print("=== Step 2: Create Initial Item ===") + + # Create test data (exact same as README) + item_data = {"Id": "123", "name": "Batman"} + + # Make POST request to create endpoint + response = api_client.post(f"{base_url}/create", json=item_data) + + # Validate response + assert response.status_code == 200, f"Create request failed with status {response.status_code}: {response.text}" + + # Check response indicates success + response_text = response.text.lower() + assert 'added' in response_text or 'created' in response_text or 'success' in response_text or 'item' in response_text, \ + f"Response should indicate item creation: {response.text}" + + print(f"✓ Item creation successful: {response.status_code}") + print(f"Created item: {item_data}") + print(f"Response: {response.text}") + print() + + +def test_03_read_item(sam_local_api, api_client): + """ + Test reading an item. + Reproduces: curl -X GET http://127.0.0.1:3000/read \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' + """ + base_url = sam_local_api + + print("=== Step 3: Read Item ===") + + # Read the item created in previous test (exact same as README) + read_data = {"Id": "123"} + + # Make GET request with JSON body (reproducing curl with -d) + response = api_client.get(f"{base_url}/read", json=read_data) + + # Validate response + assert response.status_code == 200, f"Read request failed with status {response.status_code}: {response.text}" + + # Try to parse JSON response to verify item data + try: + response_json = response.json() + + # Handle different response formats from Lambda + item_data = None + if isinstance(response_json, dict): + if 'Item' in response_json: + item_data = response_json['Item'] + elif 'Id' in response_json: + item_data = response_json + elif 'response' in response_json and 'Item' in response_json['response']: + item_data = response_json['response']['Item'] + + # Validate item data + if item_data: + # Handle DynamoDB format (with type descriptors) vs regular format + item_id = item_data.get('Id') + item_name = item_data.get('name') + + # Handle DynamoDB typed format like {"S": "value"} + if isinstance(item_id, dict) and 'S' in item_id: + item_id = item_id['S'] + if isinstance(item_name, dict) and 'S' in item_name: + item_name = item_name['S'] + + assert item_id == "123", f"Expected Id '123', got '{item_id}'" + assert item_name == "Batman", f"Expected name 'Batman', got '{item_name}'" + + print(f"✓ Item read successful: Id={item_id}, name={item_name}") + else: + # If we can't parse the structure, at least verify the data is present + response_text = response.text + assert '123' in response_text and 'Batman' in response_text, \ + f"Response should contain item data: {response.text}" + print(f"✓ Item read successful (contains expected data)") + + except (json.JSONDecodeError, KeyError): + # If JSON parsing fails, check for text indicators + response_text = response.text + assert '123' in response_text and 'Batman' in response_text, \ + f"Response should contain item data: {response.text}" + print(f"✓ Item read successful (text contains data)") + + print(f"Response: {response.text}") + print() + + +def test_04_update_item(sam_local_api, api_client): + """ + Test updating an item. + Reproduces: curl -X POST http://127.0.0.1:3000/update \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123", "name": "Robin"}' + """ + base_url = sam_local_api + + print("=== Step 4: Update Item ===") + + # Update the item (exact same as README) + update_data = {"Id": "123", "name": "Robin"} + + # Make POST request to update endpoint + response = api_client.post(f"{base_url}/update", json=update_data) + + # Validate response + assert response.status_code == 200, f"Update request failed with status {response.status_code}: {response.text}" + + # Check response indicates success + response_text = response.text.lower() + assert 'updated' in response_text or 'success' in response_text or 'modified' in response_text, \ + f"Response should indicate item update: {response.text}" + + print(f"✓ Item update successful: {response.status_code}") + print(f"Updated to: {update_data}") + print(f"Response: {response.text}") + print() + + +def test_05_check_updated_item(sam_local_api, api_client): + """ + Test reading the updated item. + Reproduces: curl -X GET http://127.0.0.1:3000/read \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' + """ + base_url = sam_local_api + + print("=== Step 5: Check Updated Item ===") + + # Read the updated item (same call as step 3) + read_data = {"Id": "123"} + + # Make GET request with JSON body + response = api_client.get(f"{base_url}/read", json=read_data) + + # Validate response + assert response.status_code == 200, f"Read updated item failed with status {response.status_code}: {response.text}" + + # Try to verify the item was updated to "Robin" + try: + response_json = response.json() + + # Handle different response formats + item_data = None + if isinstance(response_json, dict): + if 'Item' in response_json: + item_data = response_json['Item'] + elif 'Id' in response_json: + item_data = response_json + elif 'response' in response_json and 'Item' in response_json['response']: + item_data = response_json['response']['Item'] + + # Validate updated item data + if item_data: + item_name = item_data.get('name') + + # Handle DynamoDB typed format + if isinstance(item_name, dict) and 'S' in item_name: + item_name = item_name['S'] + + assert item_name == "Robin", f"Expected updated name 'Robin', got '{item_name}'" + print(f"✓ Updated item read successful: name={item_name}") + else: + # Fallback verification + response_text = response.text + assert 'Robin' in response_text, f"Response should contain updated name 'Robin': {response.text}" + print(f"✓ Updated item read successful (contains 'Robin')") + + except (json.JSONDecodeError, KeyError): + # Fallback verification + response_text = response.text + assert 'Robin' in response_text, f"Response should contain updated name 'Robin': {response.text}" + print(f"✓ Updated item read successful (text contains 'Robin')") + + print(f"Response: {response.text}") + print() + + +def test_06_delete_item(sam_local_api, api_client): + """ + Test deleting an item. + Reproduces: curl -X GET http://127.0.0.1:3000/delete \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' + """ + base_url = sam_local_api + + print("=== Step 6: Delete Item ===") + + # Delete the item (exact same as README) + delete_data = {"Id": "123"} + + # Make GET request with JSON body to delete endpoint + response = api_client.get(f"{base_url}/delete", json=delete_data) + + # Validate response + assert response.status_code == 200, f"Delete request failed with status {response.status_code}: {response.text}" + + # Check response indicates success + response_text = response.text.lower() + assert 'deleted' in response_text or 'success' in response_text or 'removed' in response_text, \ + f"Response should indicate item deletion: {response.text}" + + print(f"✓ Item deletion successful: {response.status_code}") + print(f"Deleted item: {delete_data}") + print(f"Response: {response.text}") + print() + + +def test_07_verify_item_deleted(sam_local_api, api_client): + """ + Test that the item no longer exists. + Reproduces: curl -X GET http://127.0.0.1:3000/read \ + -H 'Content-Type: application/json' \ + -d '{"Id": "123"}' + """ + base_url = sam_local_api + + print("=== Step 7: Verify Item Deleted ===") + + # Try to read the deleted item (same call as step 3 and 5) + read_data = {"Id": "123"} + + # Make GET request with JSON body + response = api_client.get(f"{base_url}/read", json=read_data) + + # Item should not be found - could be 404 or 200 with empty result + if response.status_code == 404: + print(f"✓ Item correctly deleted: {response.status_code} Not Found") + elif response.status_code == 200: + # Check if response indicates item not found + try: + response_json = response.json() + + # Verify no item data is returned + if isinstance(response_json, dict): + # Handle different response structures + item_found = False + if 'Item' in response_json and response_json['Item']: + item_found = True + elif 'response' in response_json and response_json['response'].get('Item'): + item_found = True + elif 'Id' in response_json and response_json['Id'] == '123': + item_found = True + + assert not item_found, f"Item should be deleted but was found: {response_json}" + + print(f"✓ Item correctly deleted: {response.status_code} with empty result") + + except json.JSONDecodeError: + # If response is not JSON, check for text indicators + response_text = response.text.lower() + assert 'not found' in response_text or 'empty' in response_text or len(response.text) < 50, \ + f"Response should indicate item not found: {response.text}" + print(f"✓ Item correctly deleted: text response indicates not found") + else: + pytest.fail(f"Unexpected response status for deleted item: {response.status_code}: {response.text}") + + print(f"Response: {response.text}") + print() + print("=== CRUD Sequence Complete! ===") + + +def test_08_complete_integration_cycle(sam_local_api, api_client): + """ + Test a complete CRUD cycle with a new item to verify full integration. + This test demonstrates the complete workflow in a single test. + """ + base_url = sam_local_api + test_id = "integration-456" + + print("=== Complete Integration Cycle Test ===") + print(f"Testing with ID: {test_id}") + + # Step 1: Create + print("Step 1: Create item") + create_data = {"Id": test_id, "name": "Superman", "city": "Metropolis"} + create_response = api_client.post(f"{base_url}/create", json=create_data) + assert create_response.status_code == 200, f"Create failed: {create_response.text}" + print(f"✓ Create: {create_response.status_code}") + + time.sleep(0.5) # Small delay for consistency + + # Step 2: Read + print("Step 2: Read item") + read_data = {"Id": test_id} + read_response = api_client.get(f"{base_url}/read", json=read_data) + assert read_response.status_code == 200, f"Read failed: {read_response.text}" + assert test_id in read_response.text, f"Created item not found: {read_response.text}" + print(f"✓ Read: {read_response.status_code}") + + time.sleep(0.5) + + # Step 3: Update + print("Step 3: Update item") + update_data = {"Id": test_id, "name": "Clark Kent", "city": "Smallville"} + update_response = api_client.post(f"{base_url}/update", json=update_data) + assert update_response.status_code == 200, f"Update failed: {update_response.text}" + print(f"✓ Update: {update_response.status_code}") + + time.sleep(0.5) + + # Step 4: Verify Update + print("Step 4: Verify update") + verify_response = api_client.get(f"{base_url}/read", json=read_data) + assert verify_response.status_code == 200, f"Verify failed: {verify_response.text}" + assert "Clark Kent" in verify_response.text, f"Updated name not found: {verify_response.text}" + print(f"✓ Verify: {verify_response.status_code}") + + time.sleep(0.5) + + # Step 5: Delete + print("Step 5: Delete item") + delete_response = api_client.get(f"{base_url}/delete", json=read_data) + assert delete_response.status_code == 200, f"Delete failed: {delete_response.text}" + print(f"✓ Delete: {delete_response.status_code}") + + time.sleep(0.5) + + # Step 6: Verify Deletion + print("Step 6: Verify deletion") + final_response = api_client.get(f"{base_url}/read", json=read_data) + # Should be 404 or 200 with no item + if final_response.status_code == 200: + try: + final_json = final_response.json() + # Handle different response structures for empty results + item_found = False + if isinstance(final_json, dict): + if 'Item' in final_json and final_json['Item']: + item_found = True + elif 'response' in final_json and final_json['response'].get('Item'): + item_found = True + elif 'Id' in final_json and final_json['Id'] == test_id: + item_found = True + + assert not item_found, "Item should be deleted" + except json.JSONDecodeError: + assert len(final_response.text) < 100, "Response should be minimal for deleted item" + print(f"✓ Verify deletion: {final_response.status_code}") + + print("✓ Complete integration cycle passed!") + print() + + +def test_09_error_scenarios(sam_local_api, api_client): + """ + Test error scenarios through the API Gateway integration. + """ + base_url = sam_local_api + + print("=== Error Scenarios Test ===") + + # Test 1: Read non-existent item + print("Test 1: Read non-existent item") + read_nonexistent = api_client.get(f"{base_url}/read", json={"Id": "nonexistent-999"}) + assert read_nonexistent.status_code in [200, 404], \ + f"Reading nonexistent item should return 200 or 404, got {read_nonexistent.status_code}" + print(f"✓ Read nonexistent: {read_nonexistent.status_code}") + + # Test 2: Update non-existent item + print("Test 2: Update non-existent item") + update_nonexistent = api_client.post(f"{base_url}/update", json={"Id": "nonexistent-update", "name": "Ghost"}) + assert update_nonexistent.status_code in [200, 404, 400], \ + f"Update nonexistent should return 200, 404, or 400, got {update_nonexistent.status_code}" + print(f"✓ Update nonexistent: {update_nonexistent.status_code}") + + # Test 3: Delete non-existent item + print("Test 3: Delete non-existent item") + delete_nonexistent = api_client.get(f"{base_url}/delete", json={"Id": "nonexistent-delete"}) + assert delete_nonexistent.status_code == 200, \ + f"Delete nonexistent should succeed (idempotent), got {delete_nonexistent.status_code}" + print(f"✓ Delete nonexistent: {delete_nonexistent.status_code}") + + print("✓ Error scenarios completed") + print() + + +def test_10_performance_check(sam_local_api, api_client): + """ + Test performance characteristics of the API integration. + """ + base_url = sam_local_api + + print("=== Performance Check ===") + + # Test response times for different operations + operations = [] + + # Test create performance + start_time = time.time() + create_response = api_client.post(f"{base_url}/create", json={"Id": "perf-test", "name": "Performance"}) + create_time = time.time() - start_time + operations.append(('Create', create_time * 1000, create_response.status_code)) + assert create_response.status_code == 200, f"Create failed: {create_response.status_code}" + + time.sleep(0.1) + + # Test read performance + start_time = time.time() + read_response = api_client.get(f"{base_url}/read", json={"Id": "perf-test"}) + read_time = time.time() - start_time + operations.append(('Read', read_time * 1000, read_response.status_code)) + assert read_response.status_code == 200, f"Read failed: {read_response.status_code}" + + time.sleep(0.1) + + # Test update performance + start_time = time.time() + update_response = api_client.post(f"{base_url}/update", json={"Id": "perf-test", "name": "Updated"}) + update_time = time.time() - start_time + operations.append(('Update', update_time * 1000, update_response.status_code)) + assert update_response.status_code == 200, f"Update failed: {update_response.status_code}" + + time.sleep(0.1) + + # Test delete performance + start_time = time.time() + delete_response = api_client.get(f"{base_url}/delete", json={"Id": "perf-test"}) + delete_time = time.time() - start_time + operations.append(('Delete', delete_time * 1000, delete_response.status_code)) + assert delete_response.status_code == 200, f"Delete failed: {delete_response.status_code}" + + # Report performance results + for op_name, op_time, status_code in operations: + assert op_time < 10000, f"{op_name} operation took too long: {op_time:.0f}ms" + print(f"✓ {op_name}: {op_time:.0f}ms (status: {status_code})") + + avg_time = sum(op[1] for op in operations) / len(operations) + print(f"✓ Average operation time: {avg_time:.0f}ms") + + print("✓ Performance check completed") + print() \ No newline at end of file