Build and Test an AWS SDK App with Mocks (moto, aws-sdk-client-mock, MockHandler) and Best Practices

By Ravindra Bagale, trainer · LinkedIn · Last updated: 24 October 2026

1. The big picture: the practice kitchen

Why do we need this?

A chef practises a new dish in a test kitchen with fake ingredients before cooking for paying guests. Mocks are that test kitchen: your code "calls AWS", but a fake answers instantly, costs nothing and never breaks real data.

What is it?

Write a function that takes the client as a parameter. In production pass a real client; in tests pass a mocked one. moto fakes AWS in memory, aws-sdk-client-mock stubs send() per Command, PHP MockHandler queues fake Result objects.

How to do it (step by step)

2. Build the receipts flow

Why do we need this?

Olivia snaps a A$42.50 taxi receipt. The app must store the image, record the expense and queue text reading.

What is it?

One function, three calls, clients passed in. The same shape in every language makes it easy to test.

How to do it (step by step)

1. Python.

import uuid, json
def save_receipt(s3, ddb, sqs, user, amount, body):
    rid = "r-" + uuid.uuid4().hex[:6]
    s3.put_object(Bucket="receipts-syd", Key=f"{user}/{rid}.jpg", Body=body)
    ddb.put_item(TableName="Expenses", Item={"pk": {"S": user}, "sk": {"S": rid},
        "amount": {"N": str(amount)}})
    sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=json.dumps({"rid": rid}))
    return rid
print(save_receipt(s3, ddb, sqs, "olivia", 42.50, b"jpg")[:2])

Expected output:

r-

2. Node.js.

import { PutObjectCommand } from "@aws-sdk/client-s3";
import { PutItemCommand } from "@aws-sdk/client-dynamodb";
import { SendMessageCommand } from "@aws-sdk/client-sqs";
export async function saveReceipt({ s3, ddb, sqs }, user, amount, body) {
  const rid = "r-" + Math.random().toString(16).slice(2, 8);
  await s3.send(new PutObjectCommand({ Bucket: "receipts-ngp", Key: `${user}/${rid}.jpg`, Body: body }));
  await ddb.send(new PutItemCommand({ TableName: "Expenses",
    Item: { pk: { S: user }, sk: { S: rid }, amount: { N: String(amount) } } }));
  await sqs.send(new SendMessageCommand({ QueueUrl: process.env.QUEUE_URL, MessageBody: JSON.stringify({ rid }) }));
  return rid;
}
console.log((await saveReceipt(clients, "siddharth", 780, Buffer.from("jpg"))).length);

Expected output:

8

3. PHP.

<?php
function saveReceipt($s3, $ddb, $sqs, string $user, float $amount, string $body): string {
  $rid = "r-" . bin2hex(random_bytes(3));
  $s3->putObject(["Bucket" => "receipts-ldn", "Key" => "$user/$rid.jpg", "Body" => $body]);
  $ddb->putItem(["TableName" => "Expenses", "Item" => ["pk" => ["S" => $user],
    "sk" => ["S" => $rid], "amount" => ["N" => (string)$amount]]]);
  $sqs->sendMessage(["QueueUrl" => getenv("QUEUE_URL"), "MessageBody" => json_encode(["rid" => $rid])]);
  return $rid;
}
echo strlen(saveReceipt($s3, $ddb, $sqs, "amelia", 18.20, "jpg")), PHP_EOL;

Expected output:

8

4. Verify with the CLI.

aws dynamodb query --table-name Expenses --key-condition-expression "pk = :u" --expression-attribute-values '{":u":{"S":"olivia"}}' --query Count

Expected output:

1

3. Test Python with moto

Why do we need this?

Siddharth wants tests to run on every commit, offline, in two seconds, for ₹0.

What is it?

pip install "moto[s3,dynamodb,sqs]" pytest. The @mock_aws decorator (moto 5) fakes every service; create the bucket, table and queue inside the test.

How to do it (step by step)

1. Python test.

import boto3
from moto import mock_aws
from app import save_receipt
import app
@mock_aws
def test_save_receipt():
    kw = dict(region_name="ap-southeast-2")
    s3, ddb, sqs = boto3.client("s3", **kw), boto3.client("dynamodb", **kw), boto3.client("sqs", **kw)
    s3.create_bucket(Bucket="receipts-syd", CreateBucketConfiguration={"LocationConstraint": "ap-southeast-2"})
    ddb.create_table(TableName="Expenses", BillingMode="PAY_PER_REQUEST",
        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}, {"AttributeName": "sk", "KeyType": "RANGE"}],
        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}, {"AttributeName": "sk", "AttributeType": "S"}])
    app.QUEUE_URL = sqs.create_queue(QueueName="ocr")["QueueUrl"]
    rid = save_receipt(s3, ddb, sqs, "olivia", 42.50, b"jpg")
    assert s3.get_object(Bucket="receipts-syd", Key=f"olivia/{rid}.jpg")["Body"].read() == b"jpg"
    assert ddb.get_item(TableName="Expenses", Key={"pk": {"S": "olivia"}, "sk": {"S": rid}})["Item"]["amount"]["N"] == "42.5"

Expected output:

(no output; test passes)

2. Node.js: moto as a server for any language.

// terminal: moto_server -p 5000
import { S3Client, ListBucketsCommand } from "@aws-sdk/client-s3";
const s3 = new S3Client({ region: "us-east-1", endpoint: "http://localhost:5000", forcePathStyle: true,
  credentials: { accessKeyId: "test", secretAccessKey: "test" } });
console.log((await s3.send(new ListBucketsCommand({}))).Buckets.length);

Expected output:

0

3. PHP: same server.

$s3 = new Aws\S3\S3Client(["region" => "us-east-1", "endpoint" => "http://localhost:5000",
  "use_path_style_endpoint" => true, "credentials" => ["key" => "test", "secret" => "test"]]);
echo count($s3->listBuckets()["Buckets"]), PHP_EOL;

Expected output:

0

4. Verify: run pytest.

pytest -q test_app.py

Expected output:

1 passed in 1.84s

4. Mock Node.js and PHP

Why do we need this?

Amelia in London (£18.20 lunch) wants unit tests that also check what was sent to AWS.

What is it?

npm i -D aws-sdk-client-mock. mockClient(S3Client) stubs every instance; .on(Command).resolves({...}); commandCalls() returns what was sent. In PHP push Result objects into a MockHandler and pass it as "handler".

How to do it (step by step)

1. Python: stub a single error with botocore Stubber.

from botocore.stub import Stubber
s3 = boto3.client("s3", region_name="ap-southeast-2")
with Stubber(s3) as st:
    st.add_client_error("put_object", "AccessDenied")
    try: s3.put_object(Bucket="receipts-syd", Key="x", Body=b"1")
    except s3.exceptions.ClientError as e: print(e.response["Error"]["Code"])

Expected output:

AccessDenied

2. Node.js test (node:test).

import { test } from "node:test";
import assert from "node:assert";
import { mockClient } from "aws-sdk-client-mock";
import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3";
import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
import { saveReceipt } from "./app.mjs";
const s3m = mockClient(S3Client), ddbm = mockClient(DynamoDBClient), sqsm = mockClient(SQSClient);
s3m.on(PutObjectCommand).resolves({}); ddbm.on(PutItemCommand).resolves({});
sqsm.on(SendMessageCommand).resolves({ MessageId: "m-1" });
test("saves receipt", async () => {
  await saveReceipt({ s3: new S3Client({}), ddb: new DynamoDBClient({}), sqs: new SQSClient({}) }, "siddharth", 780, "jpg");
  assert.equal(ddbm.commandCalls(PutItemCommand)[0].args[0].input.Item.amount.N, "780");
});

Expected output:

✔ saves receipt

3. PHP: MockHandler.

use Aws\MockHandler; use Aws\Result;
$mock = new MockHandler();
$mock->append(new Result([]));                      // putObject
$mock->append(new Result([]));                      // putItem
$mock->append(new Result(["MessageId" => "m-9"]));  // sendMessage
$cfg = ["region" => "eu-west-2", "handler" => $mock, "credentials" => ["key" => "k", "secret" => "s"]];
$rid = saveReceipt(new Aws\S3\S3Client($cfg), new Aws\DynamoDb\DynamoDbClient($cfg), new Aws\Sqs\SqsClient($cfg), "amelia", 18.20, "jpg");
echo count($mock), " left, rid ", strlen($rid), PHP_EOL;

Expected output:

0 left, rid 8

4. Verify: run the tests.

node --test && vendor/bin/phpunit tests

Expected output:

# pass 1
OK (1 test, 2 assertions)

5. Best practices

Why do we need this?

The app grows from 50 to 50,000 receipts a day across Sydney, Nagpur and London. Small habits now save real money and outages later.

What is it?

Create clients once and reuse them. Never hard-code keys; use roles and the default credential chain. Set timeouts and adaptive retries. Use paginators. Log RequestId on errors. Grant least-privilege IAM.

How to do it (step by step)

1. Python: timeouts and retries.

from botocore.config import Config
cfg = Config(connect_timeout=3, read_timeout=10, retries={"mode": "adaptive", "max_attempts": 5})
s3 = boto3.client("s3", region_name="ap-southeast-2", config=cfg)   # create once, reuse
print(s3.meta.config.retries["mode"])

Expected output:

adaptive

2. Node.js: reuse client outside the handler.

import { S3Client } from "@aws-sdk/client-s3";
import { NodeHttpHandler } from "@smithy/node-http-handler";
const s3 = new S3Client({ region: "ap-south-1", maxAttempts: 5, retryMode: "adaptive",
  requestHandler: new NodeHttpHandler({ connectionTimeout: 3000, requestTimeout: 10000 }) });
export const handler = async () => ({ ok: true });   // s3 reused across invocations
console.log(await s3.config.retryMode());

Expected output:

adaptive

3. PHP: log the request id on failure.

try {
  $s3->getObject(["Bucket" => "receipts-ldn", "Key" => "missing.jpg"]);
} catch (Aws\S3\Exception\S3Exception $e) {
  error_log($e->getAwsErrorCode() . " " . $e->getAwsRequestId());
  echo $e->getAwsErrorCode(), PHP_EOL;
}

Expected output:

NoSuchKey

4. Verify which identity your code uses.

aws sts get-caller-identity --query Arn --output text

Expected output:

arn:aws:sts::111122223333:assumed-role/receipts-app/i-0abc

6. Same action in Python, Node.js, Java, Go, .NET and PHP

Why do we need this?

Scripts start in the CLI. Apps call the same API through an SDK. Knowing both helps you move a tested command into real code.

What is it?

Each CLI command maps to one API call. For example aws s3api put-object ... maps to the PutObject API.

How to do it (step by step)

Same action (PutItem) from code. Every SDK reads the same profile, region and credentials as the CLI.

7. Clean up so nothing keeps costing

Why do we need this?

Lab resources left running cost money and can be a security risk.

What is it?

Delete in reverse order: children first, then parents.

How to do it (step by step)

1. Delete test bucket, table and queue (mocks need no cleanup).

aws s3 rb s3://receipts-syd --force
aws dynamodb delete-table --table-name Expenses
aws sqs delete-queue --queue-url $QUEUE_URL

Expected output:

(no output means success)

8. Interview questions and answers

Revision, comparison and interview cards

Revise this lesson in a few minutes. Save the cards on your phone or download them as PNG files. The QR code on each card opens this lesson.

More cards: Ravi Sir's Cards.