Skip to content
GenLabZ Docs

Evaluation code

Evaluation code verifies that the student created the right resources with the right configuration. It runs after the student completes a step and receives the resources they created as events, along with the validated handles from prior steps.

One evaluation file per step, located at evaluation/boto3/{objective_id}_{step_id}.py.

Function signature

async def run(session, context, logger):

Parameters

session — a boto3 Session scoped to the sandbox account.

context — a dict containing:

  • resolved — list of previously validated handles from prior steps: [{"name": "01_01_bucket", "type": "AWS::S3::Bucket", "id": "my-bucket-a3f9c2d1"}, ...]
  • events — list of resource events from the current step: [{"type": "AWS::S3::Bucket", "id": "my-bucket-a3f9c2d1"}, ...]
  • spec — the step’s expected inputs and outputs from the project spec

Events carry resource types and IDs — they do not carry handle names. Your job is to match events to expected outputs by type, then validate the resource configuration.

logger — standard Python logger.

Read-only constraint

Evaluation code must not create, modify, or delete any resources. It only reads and verifies.

Evaluation logic

For each expected output, follow this order:

  1. Check resolved — if the handle is already in context["resolved"], re-verify the resource still exists and is correctly configured
  2. Search events — if not resolved, search context["events"] for resources matching the expected output type
  3. Validate the candidate — use the service’s describe/get/head API to confirm the resource exists and meets the step’s requirements
  4. Report the result — return "found" with the resource ID, or "not_found" if no matching resource passes validation

Deep validation

Don’t just check that a resource exists — verify its configuration matches the step’s requirements. Shallow existence checks miss misconfiguration, which is the most common student error.

Examples of what to check:

ResourceWhat to validate
VPCCIDR block, DNS support, DNS hostnames
SubnetCIDR range, VPC membership, availability zone, MapPublicIpOnLaunch
Security groupVPC membership, inbound/outbound rules (ports, protocols, sources)
Route tableRoutes (destination, target), subnet associations
Load balancerScheme (internet-facing vs internal), subnets, security groups, listeners
NAT gatewaySubnet placement, EIP association, state

Matching multiple resources of the same type

When a step produces multiple outputs of the same type (e.g. two subnets), match events to outputs by validating configuration — not by position. If the spec expects a public subnet in AZ-a and a public subnet in AZ-b, check each candidate’s AZ to determine which handle it maps to.

Don’t assume event order matches output order.

Return format

return {
    "validated": [
        {
            "name": "01_01_bucket",           # handle name from spec
            "type": "AWS::S3::Bucket",        # CloudFormation resource type
            "id": "my-bucket-a3f9c2d1",       # resource ID, or None if not found
            "status": "found"                 # "found" or "not_found"
        }
    ],
    "success": True,                          # True if all outputs are "found"
    "message": "S3 bucket verified",          # human-readable summary
    "failure_context": {                      # optional, include when success is False
        "step": "01.01",
        "issue": "specific description of what's wrong",
        "hint_context": "author-written context to help generate better hints"
    }
}

success is True only when every entry in validated has status: "found".

Revalidation of prior dependencies

For steps that depend on prior outputs (declared as inputs in the spec), verify those resources still exist before checking the current step’s outputs. If a prior resource is gone, report it in validated and fail early.

Worked example — simple existence check

Spec declares: outputs: [{name: "01_01_bucket", type: "AWS::S3::Bucket"}]

from botocore.exceptions import ClientError




async def run(session, context, logger):
    s3 = session.client("s3")
    resolved = {r["name"]: r for r in context.get("resolved", [])}
    events = context.get("events", [])
    validated = []


    bucket_id = None
    if "01_01_bucket" in resolved:
        bucket_id = resolved["01_01_bucket"]["id"]
    else:
        for event in events:
            if event["type"] == "AWS::S3::Bucket" and event.get("id"):
                bucket_id = event["id"]
                break


    if not bucket_id:
        validated.append({
            "name": "01_01_bucket", "type": "AWS::S3::Bucket",
            "id": None, "status": "not_found",
        })
        return {
            "success": False, "validated": validated,
            "message": "No S3 bucket found in resolved handles or events",
            "failure_context": {
                "step": "01.01",
                "issue": "No S3 bucket detected",
                "hint_context": "Student needs to create an S3 bucket",
            },
        }


    try:
        s3.head_bucket(Bucket=bucket_id)
        validated.append({
            "name": "01_01_bucket", "type": "AWS::S3::Bucket",
            "id": bucket_id, "status": "found",
        })
    except ClientError:
        validated.append({
            "name": "01_01_bucket", "type": "AWS::S3::Bucket",
            "id": bucket_id, "status": "not_found",
        })
        return {
            "success": False, "validated": validated,
            "message": f"Bucket {bucket_id} was detected but no longer exists",
        }


    return {"success": True, "validated": validated, "message": "S3 bucket verified"}

Worked example — deep validation with prior dependency

Spec declares: inputs: ["01_01_bucket"], outputs: [{name: "01_02_policy", type: "AWS::S3::BucketPolicy"}]

import json
from botocore.exceptions import ClientError




async def run(session, context, logger):
    s3 = session.client("s3")
    resolved = {r["name"]: r for r in context.get("resolved", [])}
    events = context.get("events", [])
    validated = []


    # Revalidate prior dependency
    bucket_id = resolved.get("01_01_bucket", {}).get("id")
    if not bucket_id:
        validated.append({
            "name": "01_01_bucket", "type": "AWS::S3::Bucket",
            "id": None, "status": "not_found",
        })
        validated.append({
            "name": "01_02_policy", "type": "AWS::S3::BucketPolicy",
            "id": None, "status": "not_found",
        })
        return {
            "success": False, "validated": validated,
            "message": "Prior dependency 01_01_bucket not resolved",
            "failure_context": {
                "step": "01.02", "issue": "Bucket dependency missing",
                "hint_context": "Complete step 01.01 first.",
            },
        }


    try:
        s3.head_bucket(Bucket=bucket_id)
    except ClientError:
        validated.append({
            "name": "01_01_bucket", "type": "AWS::S3::Bucket",
            "id": bucket_id, "status": "not_found",
        })
        validated.append({
            "name": "01_02_policy", "type": "AWS::S3::BucketPolicy",
            "id": None, "status": "not_found",
        })
        return {
            "success": False, "validated": validated,
            "message": f"Prior bucket {bucket_id} no longer exists",
        }


    # Deep validation: check the policy grants public read
    try:
        policy_response = s3.get_bucket_policy(Bucket=bucket_id)
        policy = json.loads(policy_response["Policy"])


        has_public_read = False
        for stmt in policy.get("Statement", []):
            if (stmt.get("Effect") == "Allow"
                    and stmt.get("Principal") in ("*", {"AWS": "*"})
                    and "s3:GetObject" in stmt.get("Action", [])):
                has_public_read = True
                break


        if not has_public_read:
            validated.append({
                "name": "01_02_policy", "type": "AWS::S3::BucketPolicy",
                "id": bucket_id, "status": "not_found",
            })
            return {
                "success": False, "validated": validated,
                "message": "Bucket policy exists but does not grant public read access",
                "failure_context": {
                    "step": "01.02",
                    "issue": "Policy missing s3:GetObject for public principal",
                    "hint_context": "The bucket policy needs an Allow statement for s3:GetObject with Principal: *",
                },
            }


        validated.append({
            "name": "01_02_policy", "type": "AWS::S3::BucketPolicy",
            "id": bucket_id, "status": "found",
        })


    except ClientError as e:
        code = e.response["Error"]["Code"]
        if code == "NoSuchBucketPolicy":
            validated.append({
                "name": "01_02_policy", "type": "AWS::S3::BucketPolicy",
                "id": None, "status": "not_found",
            })
            return {
                "success": False, "validated": validated,
                "message": "No bucket policy found on the bucket",
                "failure_context": {
                    "step": "01.02",
                    "issue": "Bucket exists but has no bucket policy attached",
                    "hint_context": "Student needs to add a bucket policy to the S3 bucket",
                },
            }
        else:
            validated.append({
                "name": "01_02_policy", "type": "AWS::S3::BucketPolicy",
                "id": None, "status": "not_found",
            })
            return {
                "success": False, "validated": validated,
                "message": f"Error checking bucket policy: {code}",
            }


    return {"success": True, "validated": validated, "message": "Bucket policy verified"}