test_lambda_handler.py

"""End-to-end smoke test using moto for DynamoDB. No AWS calls."""
import json
import os
import sys
from pathlib import Path

os.environ["DJANGO_SETTINGS_MODULE"] = "config.settings"
os.environ["DJANGO_DEBUG"] = "True"
os.environ["TODO_TABLE_NAME"] = "workshop-todos-test"
os.environ.setdefault("AWS_DEFAULT_REGION", "us-west-2")
os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")

sys.path.insert(0, str(Path(__file__).parent / "todo_api"))

import boto3  # noqa: E402
from moto import mock_aws  # noqa: E402


class FakeContext:
    function_name = "workshop-todo-api"
    function_version = "$LATEST"
    invoked_function_arn = "arn:aws:lambda:ap-southeast-1:000000000000:function:workshop-todo-api"
    memory_limit_in_mb = 1024
    aws_request_id = "test-req-1"
    log_group_name = "/aws/lambda/workshop-todo-api"
    log_stream_name = "2026/05/12/[$LATEST]abcdef"

    def get_remaining_time_in_millis(self):
        return 30000


def make_event(method, path, body=None):
    # split path?query
    if "?" in path:
        raw_path, raw_qs = path.split("?", 1)
    else:
        raw_path, raw_qs = path, ""
    headers = {"host": "abc.execute-api.us-west-2.amazonaws.com"}
    raw_body = ""
    if body is not None:
        raw_body = json.dumps(body)
        headers["content-type"] = "application/json"
        headers["content-length"] = str(len(raw_body))
    return {
        "version": "2.0",
        "routeKey": "$default",
        "rawPath": raw_path,
        "rawQueryString": raw_qs,
        "headers": headers,
        "requestContext": {
            "accountId": "000000000000",
            "apiId": "abc",
            "domainName": "abc.execute-api.ap-southeast-1.amazonaws.com",
            "http": {
                "method": method,
                "path": raw_path,
                "protocol": "HTTP/1.1",
                "sourceIp": "127.0.0.1",
                "userAgent": "test-client",
            },
            "requestId": "test-req-1",
            "routeKey": "$default",
            "stage": "$default",
            "time": "12/May/2026:00:00:00 +0000",
            "timeEpoch": 1747008000000,
        },
        "body": raw_body or None,
        "isBase64Encoded": False,
    }


def run(label, handler, event):
    print(f"\n=== {label} ===")
    resp = handler(event, FakeContext())
    print(f"status: {resp['statusCode']}")
    print(f"body:   {resp.get('body', '')[:300]}")
    return resp


def assert_eq(label, got, want):
    ok = got == want
    print(f"  [{'OK' if ok else 'FAIL'}] {label}: got={got!r} want={want!r}")
    if not ok:
        sys.exit(1)


@mock_aws
def main():
    ddb = boto3.client("dynamodb")
    ddb.create_table(
        TableName=os.environ["TODO_TABLE_NAME"],
        AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        BillingMode="PAY_PER_REQUEST",
    )
    ddb.get_waiter("table_exists").wait(TableName=os.environ["TODO_TABLE_NAME"])

    # Import inside @mock_aws so the boto3 client points at the mocked endpoint.
    from config.handler import handler

    # Pagination check at end uses these.
    r = run("GET / (HTML UI)", handler, make_event("GET", "/"))
    assert_eq("status", r["statusCode"], 200)
    ctype = next((v for k, v in r.get("headers", {}).items() if k.lower() == "content-type"), "")
    assert "text/html" in ctype, f"expected text/html, got {ctype!r}"
    assert "Workshop Todos" in r["body"], "UI template body missing"
    assert "alpinejs" in r["body"], "Alpine.js CDN missing from UI"

    r = run("GET /healthz", handler, make_event("GET", "/healthz"))
    assert_eq("status", r["statusCode"], 200)
    assert json.loads(r["body"]) == {"ok": True}

    r = run("GET /api/todos/ (empty)", handler, make_event("GET", "/api/todos/"))
    assert_eq("status", r["statusCode"], 200)
    page = json.loads(r["body"])
    assert page == {"items": [], "next_cursor": None}, page

    r = run("POST /api/todos/", handler,
            make_event("POST", "/api/todos/", {"title": "via DynamoDB", "done": False}))
    assert_eq("status", r["statusCode"], 201)
    created = json.loads(r["body"])
    assert created["title"] == "via DynamoDB"
    assert "id" in created and "created_at" in created
    todo_id = created["id"]

    r = run("GET /api/todos/ (after create)", handler, make_event("GET", "/api/todos/"))
    page = json.loads(r["body"])
    assert len(page["items"]) == 1 and page["items"][0]["id"] == todo_id, page
    assert page["next_cursor"] is None, page

    r = run(f"GET /api/todos/{todo_id}/", handler, make_event("GET", f"/api/todos/{todo_id}/"))
    assert_eq("status", r["statusCode"], 200)
    assert json.loads(r["body"])["title"] == "via DynamoDB"

    r = run("PATCH (toggle done)", handler,
            make_event("PATCH", f"/api/todos/{todo_id}/", {"done": True}))
    assert_eq("status", r["statusCode"], 200)
    assert json.loads(r["body"])["done"] is True

    r = run("GET unknown", handler, make_event("GET", "/api/todos/does-not-exist/"))
    assert_eq("status", r["statusCode"], 404)

    r = run("DELETE", handler, make_event("DELETE", f"/api/todos/{todo_id}/"))
    assert_eq("status", r["statusCode"], 204)

    r = run("GET /api/todos/ (after delete)", handler, make_event("GET", "/api/todos/"))
    assert json.loads(r["body"]) == {"items": [], "next_cursor": None}

    # Pagination: create 3 todos, GET with limit=2 yields 2 + cursor, page 2 yields 1 + no cursor
    for i in range(3):
        handler(make_event("POST", "/api/todos/", {"title": f"page-test-{i}", "done": False}), FakeContext())

    r1 = run("GET ?limit=2 (page 1)", handler, make_event("GET", "/api/todos/?limit=2"))
    page1 = json.loads(r1["body"])
    assert len(page1["items"]) == 2, page1
    assert page1["next_cursor"], "expected next_cursor on page 1"

    from urllib.parse import quote
    r2 = run(f"GET ?limit=2&cursor=... (page 2)", handler,
             make_event("GET", f"/api/todos/?limit=2&cursor={quote(page1['next_cursor'])}"))
    page2 = json.loads(r2["body"])
    assert len(page2["items"]) == 1, page2
    assert page2["next_cursor"] is None, page2

    print("\nAll smoke tests passed.")


if __name__ == "__main__":
    main()