AWS Builder Center

Builder Center

Testing AI-Generated Code in Seconds with Code Interpreter: A PySpark Case Study (and Beyond)

Using Amazon Bedrock AgentCore Code Interpreter as a universal validation sandbox — from PySpark to SQL to Lambda handlers

Mike Brouwers

AWS Solutions Architect
Part 2 of 2: Using Amazon Bedrock AgentCore Code Interpreter as a universal validation sandbox — from PySpark to SQL to Lambda handlers

The Problem: Static Analysis Can't Tell You If Code Produces Correct Output

In  Part 1 , we covered how to scope what an AI agent is allowed to decide — constrain its authority, contain its output in templates, validate structure with AST analysis, and gate deployment with quality scores. Those layers catch a lot of bugs. But they share a fundamental limitation.
Static analysis tells you the code is well-formed. It doesn't tell you the code produces correct output.
The only way to know if AI-generated code actually works is to run it. And here's the challenge: the generated code's native runtime is often expensive, slow, or both. Spark clusters take minutes to start. Deploying CDK stacks costs real money. Executing untested SQL against production databases is off the table. You need a way to validate logic before committing to the expensive execution path.

The Solution: Code Interpreter as Your Fast Validation Layer

Amazon Bedrock AgentCore's Code Interpreter is a Python sandbox that starts in seconds, costs fractions of a cent, and runs in complete isolation. For any AI-generated code whose logic can be expressed or validated in Python — PySpark transforms, SQL queries, Lambda handlers, IaC constructs — Code Interpreter is your fast validation layer.
The pattern is:
  1. Generate code (with your agent)
  1. Execute or validate it in Code Interpreter (with synthetic data, in seconds)
  1. Gate deployment on the results (auto-approve, retry, or escalate)
You don't need the native runtime to catch most logic bugs. You need a fast, cheap sandbox that preserves enough semantics to surface the failures that matter.
In this article, we'll walk through a detailed implementation — converting PySpark transforms to pandas for testing in Code Interpreter — then show how the same pattern applies to SQL, Lambda handlers, and CDK constructs without any conversion at all.


Case Study: PySpark Without Spark

Our system generates PySpark transformation code that runs as AWS Glue jobs. Consider this AI-generated transform block:
Python
df_transformed = df.withColumn("amount", col("amount").cast(IntegerType()))
df_transformed = df_transformed.filter(col("amount").isNotNull())
df_transformed = df_transformed.withColumn("name", lower(col("name")))
This passes every static check from Part 1. Valid syntax. Correct function-call style. No anti-patterns. But does it work? Does casting  "$45.99"  to  IntegerType  produce  null  or  45 ? Does the null filter then drop those rows? Is that what the user intended?
Running this natively requires Apache Spark — a JVM-based cluster that takes 3-8 minutes to start. In a pipeline that generates code in seconds, that's unacceptable.
Code Interpreter has pandas, NumPy, and the Python standard library. It does not have PySpark. But here's the insight: our agent only generates a narrow subset of PySpark operations (thanks to the scope constraints from Part 1). That subset has near-direct pandas equivalents. We can convert the generated PySpark to pandas, run it in Code Interpreter with synthetic data, and validate the output — all in under 5 seconds.


The Approach: Transpile PySpark to Pandas with Ordered Regex

Let's be direct: converting PySpark to pandas using regex substitutions sounds like a terrible idea. PySpark and pandas have different semantics, different null handling, different type systems, and different execution models (lazy vs. eager).
But remember our architectural advantage from Part 1: the agent only generates transform blocks. Not full PySpark applications. Not Spark SQL. Not RDD operations. Just a narrow subset of DataFrame operations:  .withColumn() ,  .select() ,  .filter() ,  .cast() ,  .withColumnRenamed() , and a whitelist of functions from  pyspark.sql.functions .
That narrow scope is what makes regex conversion viable. We're not transpiling arbitrary PySpark — we're converting a known, constrained subset where PySpark and pandas have direct equivalents.
Key insight: The scope constraint from Part 1 (Layer 1: constrain the agent's decision space) directly enables the testing strategy in Part 2. Architectural decisions compound.


The Conversion Engine: 25 Ordered Pattern Rules

The converter is a class with an ordered list of  (regex_pattern, replacement)  tuples. Order matters — more specific patterns must fire before generic ones to prevent partial matches.
Python
class PySpark2PandasConverter:
    """Convert a PySpark transform block to pandas equivalent code."""

    PATTERNS = [
        # --- Strip PySpark imports (template provides them) ---
        (r"from pyspark\.sql\.functions import .*\n?", ""),
        (r"from pyspark\.sql\.types import .*\n?", ""),

        # --- withColumnRenamed ---
        (r'\.withColumnRenamed\(\s*"(\w+)"\s*,\s*"(\w+)"\s*\)',
         r'.rename(columns={"\1": "\2"})'),

        # --- col().cast() for various types ---
        (r'col\(\s*"(\w+)"\s*\)\.cast\(\s*IntegerType\(\)\s*\)',
         r'pd.to_numeric(df["\1"], errors="coerce").astype("Int64")'),

        (r'col\(\s*"(\w+)"\s*\)\.cast\(\s*DoubleType\(\)\s*\)',
         r'pd.to_numeric(df["\1"], errors="coerce")'),

        (r'col\(\s*"(\w+)"\s*\)\.cast\(\s*StringType\(\)\s*\)',
         r'df["\1"].astype(str)'),

        # --- String functions ---
        (r'trim\(\s*col\(\s*"(\w+)"\s*\)\s*\)',
         r'df["\1"].str.strip()'),

        (r'lower\(\s*col\(\s*"(\w+)"\s*\)\s*\)',
         r'df["\1"].str.lower()'),

        # --- Date/timestamp ---
        (r'to_date\(\s*col\(\s*"(\w+)"\s*\).*?\)',
         r'pd.to_datetime(df["\1"], errors="coerce").dt.date'),

        # --- Null filters ---
        (r'\.filter\(\s*col\(\s*"(\w+)"\s*\)\.isNotNull\(\)\s*\)',
         r'[df["\1"].notna()]'),

        # --- when().otherwise() ---
        (r'when\(\s*col\(\s*"(\w+)"\s*\)\s*==\s*"([^"]+)"\s*,\s*"([^"]+)"\s*\)'
         r'\.otherwise\(\s*"([^"]+)"\s*\)',
         r'np.where(df["\1"] == "\2", "\3", "\4")'),

        # --- col("x") references → df["x"] (must be LAST) ---
        (r'col\(\s*"(\w+)"\s*\)', r'df["\1"]'),
    ]
The full converter has ~25 patterns covering casts, string operations, date parsing, null handling, regex replacement, concat, literal values, select/drop, and conditional logic. Here's the complete mapping:
PySpark
Pandas Equivalent
 .withColumnRenamed("a", "b") 
 .rename(columns={"a": "b"}) 
 col("x").cast(IntegerType()) 
 pd.to_numeric(df["x"], errors="coerce").astype("Int64") 
 col("x").cast(DoubleType()) 
 pd.to_numeric(df["x"], errors="coerce") 
 col("x").cast(StringType()) 
 df["x"].astype(str) 
 trim(col("x")) 
 df["x"].str.strip() 
 lower(col("x")) 
 df["x"].str.lower() 
 upper(col("x")) 
 df["x"].str.upper() 
 to_date(col("x")) 
 pd.to_datetime(df["x"], errors="coerce").dt.date 
 to_timestamp(col("x")) 
 pd.to_datetime(df["x"], errors="coerce") 
 .filter(col("x").isNotNull()) 
 [df["x"].notna()] 
 .filter(col("x").isNull()) 
 [df["x"].isna()] 
 regexp_replace(col("x"), "pat", "rep") 
 df["x"].str.replace(r"pat", "rep", regex=True) 
 concat_ws(" ", col("a"), col("b")) 
 df["a"].astype(str) + " " + df["b"].astype(str) 
 when(col("x") == "a", "b").otherwise("c") 
 np.where(df["x"] == "a", "b", "c") 
 lit("value") 
 "value" 
 .drop("col") 
 .drop(columns=["col"]) 
 df = df.withColumn("x", expr) 
 df["x"] = expr 


The Unconvertible Line Problem

Here's where the approach gets pragmatic. Not every PySpark operation has a regex-convertible pandas equivalent. Joins, aggregations, UDFs, RDD operations, and window functions can't be mechanically translated with pattern matching.
Our solution: comment them out instead of failing.
Python
_PYSPARK_RESIDUALS = re.compile(
    r"\b(?:col\(|F\.|functions\.|"
    r"StringType|IntegerType|LongType|DoubleType|"
    r"\.groupBy\(|\.agg\(|\.join\(|\.union\(|"
    r"\.crossJoin\(|\.repartition\(|"
    r"udf\(|\.rdd\.|spark\.)"
)

@classmethod
def _wrap_unconverted(cls, code: str) -> str:
    """Comment out lines that still contain PySpark constructs."""
    lines = code.split("\n")
    result = []
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            result.append(line)
            continue
        if cls._PYSPARK_RESIDUALS.search(stripped):
            result.append(f"# UNCONVERTED: {line}")
        else:
            result.append(line)
    return "\n".join(result)
After all 25 conversion patterns run, the converter scans every remaining line for PySpark residuals. Any line that still contains PySpark-specific constructs gets prefixed with  # UNCONVERTED: .
This means the pandas code can always execute. Unconverted lines become comments — they don't crash the sandbox. The test validates the portions of the transform that could be converted rather than failing entirely because one line uses an unsupported operation.
Why this is better than failing: If the agent generates 10 lines and 9 convert cleanly, testing those 9 lines is vastly more valuable than testing nothing. The one unconverted line surfaces in the test report as a flag for human review — a known gap, not a silent miss.


The Test Lifecycle: Three Phases in a Code Interpreter Session

The full test runs inside a single AgentCore Code Interpreter session. Three code snippets execute sequentially in the same session, sharing state:
Plain text
┌─────────────────────────────────────────────────────────────┐
│  Code Interpreter Session (AgentCore)                        │
│                                                              │
│  Phase 1: Generate Test Data                                 │
│  ┌────────────────────────────────────────────────────────┐  │
│  │ Build a pandas DataFrame from source_schema             │  │
│  │ 100 rows, realistic types, ~5% injected nulls           │  │
│  │ Result: `df` variable in session scope                  │  │
│  └────────────────────────────────────────────────────────┘  │
│                          │                                   │
│                          ▼                                   │
│  Phase 2: Execute Converted Transform                        │
│  ┌────────────────────────────────────────────────────────┐  │
│  │ PySpark2PandasConverter.convert(transform_block)        │  │
│  │ Execute pandas equivalent in session                    │  │
│  │ Result: `df_transformed` variable in session scope      │  │
│  └────────────────────────────────────────────────────────┘  │
│                          │                                   │
│                          ▼                                   │
│  Phase 3: Validate Output                                    │
│  ┌────────────────────────────────────────────────────────┐  │
│  │ Check record count, null distribution, column presence  │  │
│  │ Compute quality scores: completeness, uniqueness,       │  │
│  │   schema compliance                                     │  │
│  │ Result: structured JSON validation report               │  │
│  └────────────────────────────────────────────────────────┘  │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Phase 1: Synthetic Test Data Generation

We generate test data dynamically from the source schema — not from the actual production data. This is deliberate: we want to test the transform logic, not the data itself.
Python
def _build_test_data_code(self, source_schema: dict) -> str:
    """Build Python code that creates a test DataFrame from schema."""
    fields = source_schema.get("properties", {})
    col_defs = []

    for name, meta in list(fields.items())[:20]:
        dtype = meta.get("type", "string")
        if dtype in ("integer", "int", "long"):
            col_defs.append(f'    "{name}": [i for i in range(num_rows)]')
        elif dtype in ("float", "double", "decimal"):
            col_defs.append(f'    "{name}": [round(i * 1.1, 2) for i in range(num_rows)]')
        else:
            col_defs.append(f'    "{name}": [f"val_{{i}}" for i in range(num_rows)]')

    return f"""
import pandas as pd, numpy as np, json, random
num_rows = 100
data = {{
{chr(10).join(col_defs)}
}}
df = pd.DataFrame(data)
# Inject ~5% nulls to test null handling
for c in df.columns:
    mask = [random.random() < 0.05 for _ in range(num_rows)]
    df.loc[mask, c] = None
print("GENERATED:" + json.dumps({{"record_count": len(df), "columns": list(df.columns)}}))
"""
Key design decisions:
  • 100 rows — enough to exercise transforms, small enough to run in milliseconds
  • Type-aware generation — integers get sequences, floats get decimals, strings get predictable patterns
  • 5% null injection — tests whether the transform handles nulls gracefully or crashes
  • Cap at 20 columns — prevents extremely wide schemas from bloating the session

Phase 2: Execute the Converted Transform

The PySpark transform block is converted to pandas and sent to the session:
Python
# In the test runner:
pandas_code = PySpark2PandasConverter.convert(transform_block)
self._invoke(pandas_code)  # Execute in Code Interpreter session
Because Phase 1 left  df  in session scope, and the converted code operates on  df  and produces  df_transformed , the pandas equivalent runs against the synthetic test data. If the conversion is correct and the logic is sound,  df_transformed  will be a valid DataFrame in session scope when Phase 3 begins.
If execution fails — a TypeError, a KeyError, a conversion gap — Phase 3 still runs and reports the error. The test doesn't crash; it reports what went wrong.

Phase 3: Output Validation

The validation code checks the output DataFrame against the target schema and computes quality metrics:
Python
record_count = len(df_transformed)
null_counts = {c: int(df_transformed[c].isna().sum()) for c in df_transformed.columns}
duplicates = int(df_transformed.duplicated().sum())
missing_columns = [c for c in expected_columns if c not in df_transformed.columns]

completeness = (1 - sum(null_counts.values()) / (record_count * len(df_transformed.columns))) * 100
uniqueness = (1 - duplicates / record_count) * 100
schema_compliance = (1 - len(missing_columns) / len(expected_columns)) * 100
overall_quality = (completeness + uniqueness + schema_compliance) / 3
The output is a structured JSON report with four scores:
  • Completeness — what percentage of cells are non-null?
  • Uniqueness — are there unexpected duplicates?
  • Schema compliance — does the output have the expected columns?
  • Overall quality — weighted average of the three


What This Catches That Static Analysis Misses

Bug Class
Static Analysis (Part 1)
Runtime Testing (Part 2)
col("x").lower() anti-pattern
Catches it
Would also catch (runtime error)
Cast drops all values to null
Cannot detect
Catches it (completeness score drops)
Filter removes all rows
Cannot detect
Catches it (record count = 0)
Column rename creates wrong name
Cannot detect
Catches it (schema compliance fails)
Transform produces duplicates
Cannot detect
Catches it (uniqueness score drops)
Missing column in output
Cannot detect
Catches it (missing_columns list)
TypeError from incompatible cast
Cannot detect
Catches it (execution failure)
The pattern is clear: static analysis catches structural bugs (wrong syntax, wrong API usage). Runtime testing catches semantic bugs (correct structure but wrong behavior). You need both.


Why Not Just Run PySpark Locally?

Fair question. Here's the tradeoff matrix:
Approach
Startup Time
Cost
Fidelity
Catches
Spark cluster (Glue/EMR)
3-8 minutes
$$
Perfect
Everything
Local Spark (Docker)
30-60 seconds
$
High
Most things
Pandas conversion (our approach)
2-5 seconds
Fractions of a cent
Moderate
Logic + data quality bugs
Static analysis only
Milliseconds
Free
Low
Structural bugs only
We chose the pandas approach because our pipeline needs to validate code before human review and before deployment. The validation is part of an automated Step Functions pipeline where seconds matter. If validation took 5 minutes, it would negate the value of automated generation.
The moderate fidelity is acceptable because:
  1. Our constrained scope (15 lines of .withColumn/.filter/.cast) has near-perfect conversion fidelity for the operations we allow
  1. Any unconvertible lines are explicitly flagged — we know what we didn't test
  1. The real Spark execution happens in the next pipeline step (Glue job creation + execution) with a result validation gate after it
The pandas test isn't the only validation. It's the fast validation that gates whether we proceed to the expensive validation.


The Feedback Loop: When Tests Fail, Regenerate

Test results feed back into code generation. If the pandas execution reveals issues, the system determines whether regeneration is warranted:
Python
def analyze_test_results_for_regeneration(
    test_status, tests_passed, tests_failed,
    coverage_percent, review_result, regeneration_attempt
):
    regeneration_needed = False

    if test_status == "failed" or tests_failed > 0:
        if regeneration_attempt < 3:  # Max 3 iterations
            regeneration_needed = True

    if coverage_percent < 70:
        if regeneration_attempt < 3:
            regeneration_needed = True

    return {
        "regeneration_needed": regeneration_needed,
        "max_attempts_reached": regeneration_attempt >= 3,
    }
When regeneration triggers, the feedback — what failed, why, which tests — is stored in DynamoDB. The code generation agent (from Part 1) retrieves this feedback on its next attempt, giving it specific context about what went wrong:
Plain text
Generate code (attempt 1)
    │
    ▼
Convert PySpark → pandas → execute in Code Interpreter
    │
    ▼ (test fails: schema_compliance = 60%)
    │
Store feedback: "missing columns: ['order_date', 'total_amount']"
    │
    ▼
Generate code (attempt 2, with feedback context)
    │
    ▼
Convert → execute → validate
    │
    ▼ (test passes: overall_quality = 92%)
    │
Proceed to deployment
The cap at 3 attempts prevents infinite loops. If three attempts can't produce passing code, the system escalates to human review rather than burning compute on a problem the agent can't solve.


Dual-Layer Testing: Static Analysis + Execution

Our complete testing step combines both approaches in sequence:
  1. Phase 1: Static analysis via Strands Agent. A second AI agent reviews the generated code for correctness, best practices, error handling, and security. This produces a structured report with pass/fail categories.
  1. Phase 2: Runtime execution via Code Interpreter. The PySpark-to-pandas conversion runs in a sandbox with synthetic data. This produces quality scores and data validation results.
The two phases produce complementary signals that are combined into a single quality assessment:
Python
if execution_results.get("execution_successful"):
    # Execution success boosts the overall assessment
    test_status = "passed"
    tests_passed += 3  # Credit for runtime validation
    tests_run += 3
    # Blend execution quality with static analysis coverage
    coverage_percent = (static_coverage + execution_quality) / 2
Successful Code Interpreter execution doesn't just add a pass/fail — it boosts the confidence score. A transform that runs correctly against synthetic data is meaningfully more trustworthy than one that only passes static review.


The Honest Tradeoffs

This approach has real limitations. Being transparent about them is important:

1. Semantic gaps in conversion

PySpark and pandas handle nulls differently. PySpark has typed nulls; pandas has  NaN  for floats and  None  for objects. A transform that works in pandas might behave differently in Spark with nullable  IntegerType  columns. We mitigate this by using pandas nullable types ( "Int64"  instead of  int64 ) but the gap exists.

2. Scale-dependent bugs are invisible

A transform that works on 100 rows might OOM on 100 million. Partition skew, broadcast join limits, and shuffle failures can't be caught in pandas. These surface in the actual Glue execution step downstream.

3. The UNCONVERTED escape hatch

Any line marked  # UNCONVERTED  is untested. If the agent generates complex operations (joins, window functions, UDFs) that fall outside our whitelisted scope, those operations pass through without runtime validation. The mitigation: our constrained prompt (Part 1, Layer 1) makes this rare — about 5% of generated transforms contain unconvertible lines.

4. Regex maintenance cost

Each new PySpark pattern the agent learns to generate requires a new conversion rule. The regex list grows over time. Currently at 25 patterns — manageable, but it's technical debt that scales with model capability.
Despite these limitations: the pandas conversion catches ~80% of the semantic bugs we'd otherwise only discover after a 5-minute Spark job fails. In a pipeline optimized for speed-to-deployment, that's a massive win.


The Complete Guardrails Picture (Parts 1 + 2)

Combining both articles, here's the full validation pipeline for AI-generated code:
Plain text
┌─────────────────────────────────────────────────────────────────────┐
│ GENERATION (Part 1)                                                  │
│                                                                      │
│  Constrained Prompt → AI Agent → Template Assembly                   │
│       │                                                              │
│       └── AST validation → anti-pattern check → quality score        │
│                                                                      │
├─────────────────────────────────────────────────────────────────────┤
│ TESTING (Part 2)                                                     │
│                                                                      │
│  Static Analysis (AI agent review of code quality)                   │
│       │                                                              │
│       ▼                                                              │
│  PySpark → Pandas conversion                                         │
│       │                                                              │
│       ▼                                                              │
│  Code Interpreter execution (synthetic data, quality scores)         │
│       │                                                              │
│       ▼                                                              │
│  Combined quality assessment → regenerate / approve / escalate       │
│                                                                      │
├─────────────────────────────────────────────────────────────────────┤
│ DEPLOYMENT                                                           │
│                                                                      │
│  Create Glue job → Execute on Spark cluster → Validate results       │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘
Three layers of increasing fidelity:
  1. Static (milliseconds) — catches structural bugs
  1. Sandbox (seconds) — catches semantic/logic bugs
  1. Production (minutes) — catches scale and environment bugs
Each layer is a gate. If code fails at Layer 1, it never reaches Layer 2. If it fails at Layer 2, it never burns expensive cluster time at Layer 3. The cheapest validation runs first.


Code Interpreter as a Universal Test Sandbox

We've been describing our PySpark-to-pandas conversion as if it's a one-off trick for a niche framework. But step back and look at what's actually happening: we're using AgentCore Code Interpreter as a fast, cheap, isolated execution environment to validate AI-generated code before committing to expensive deployment.
That pattern isn't specific to PySpark. Code Interpreter is a Python sandbox with pandas, NumPy, and the standard library. For any AI-generated code that can execute in Python — or whose logic can be validated in Python — Code Interpreter is already your lightweight test environment.

SQL Generation: Execute Against SQLite In-Session

Python's  sqlite3  is in the standard library. If you're generating PostgreSQL or Redshift SQL, you can validate the logic directly in Code Interpreter:
Python
import sqlite3, json

# Create in-memory database with synthetic data from schema
conn = sqlite3.connect(":memory:")
conn.execute("""
    CREATE TABLE orders (
        order_id INTEGER, customer_id INTEGER,
        amount REAL, status TEXT, created_at TEXT
    )
""")
# Insert synthetic rows
for i in range(100):
    conn.execute(
        "INSERT INTO orders VALUES (?, ?, ?, ?, ?)",
        (i, i % 20, round(i * 9.99, 2), 
         "active" if i % 3 else "cancelled", 
         f"2026-01-{(i%28)+1:02d}")
    )

# Execute the AI-generated SQL
generated_sql = """
    SELECT customer_id, SUM(amount) as total_spend, COUNT(*) as order_count
    FROM orders
    WHERE status = 'active'
    GROUP BY customer_id
    HAVING total_spend > 50
"""
result = conn.execute(generated_sql).fetchall()
print(json.dumps({"row_count": len(result), "columns": 3, "execution": "success"}))
No conversion needed. No lighter database to find. Code Interpreter is the lighter database. You'll miss Redshift-specific functions ( LISTAGG , window frames), but you'll catch JOIN logic errors, GROUP BY mistakes, WHERE clause bugs, and type mismatches — which is where most AI-generated SQL fails.

Lambda Handler Generation: Direct Invocation In-Session

If your agent generates Python Lambda handlers, you can test them directly in Code Interpreter — no SAM, no Docker, no deployment:
Python
import json

# The AI-generated handler
def lambda_handler(event, context):
    body = json.loads(event.get("body", "{}"))
    user_id = body.get("user_id")
    if not user_id:
        return {"statusCode": 400, "body": json.dumps({"error": "user_id required"})}
    # ... business logic ...
    return {"statusCode": 200, "body": json.dumps({"user_id": user_id, "status": "processed"})}

# Test with synthetic events
test_cases = [
    {"body": json.dumps({"user_id": "u-123"})},           # Happy path
    {"body": json.dumps({})},                              # Missing field
    {"body": "invalid json"},                              # Malformed input
    {},                                                     # Empty event
]

results = []
for event in test_cases:
    try:
        resp = lambda_handler(event, None)
        results.append({"status": resp["statusCode"], "pass": resp["statusCode"] in [200, 400]})
    except Exception as e:
        results.append({"status": "error", "pass": False, "error": str(e)})

print(json.dumps({"tests_run": len(results), "tests_passed": sum(r["pass"] for r in results)}))
You're testing the actual generated code against realistic events. No mocking frameworks, no infrastructure. If the handler crashes on malformed input or forgets to handle edge cases, you catch it in seconds.

CDK Construct Validation: Programmatic Synthesis In-Session

Code Interpreter supports  install_packages . You can install  aws-cdk-lib  and synthesize generated constructs to validate they produce valid CloudFormation — without deploying anything:
Python
# After: install_packages(["aws-cdk-lib", "constructs"])
from aws_cdk import App, Stack, assertions
from aws_cdk import aws_s3 as s3, aws_iam as iam
import constructs

# The AI-generated construct
app = App()
stack = Stack(app, "TestStack")

# ... inject generated CDK code here ...
bucket = s3.Bucket(stack, "DataBucket", encryption=s3.BucketEncryption.S3_MANAGED)

# Validate the synthesized template
template = assertions.Template.from_stack(stack)

# Assert security properties exist
template.has_resource_properties("AWS::S3::Bucket", {
    "BucketEncryption": {"ServerSideEncryptionConfiguration": [{}]}
})

print(json.dumps({"synthesis": "success", "resources": len(template.to_json()["Resources"])}))
This catches the most common AI-generated IaC failures: missing encryption, overly permissive access, invalid property combinations — all without touching your AWS account.

The Unified Pattern

Code Interpreter isn't just a sandbox for our specific PySpark use case. It's the universal "fast, cheap, isolated" test environment for any AI-generated code. The pattern is: generate code → execute in Code Interpreter with synthetic data → validate the output → gate deployment on results.
What makes this work:
  • Sessions start in seconds — no cold start on a cluster or container
  • Sessions are isolated — each test runs in a clean environment, no cross-contamination
  • Python is the lingua franca — if you can express your validation logic in Python, it runs here
  • Packages are installable — need  sqlite3 ? It's built in. Need  aws-cdk-lib ? Install it. Need  boto3 ? Already there.
  • Cost is negligible — fractions of a cent per session, versus dollars for Spark clusters or deployed infrastructure
For any domain where you're generating code with AI, ask: "Can I validate this logic in a Code Interpreter session?" If yes, you have a sub-10-second test gate that catches most semantic bugs before you commit to expensive downstream execution.


Key Takeaways

  1. Scope constraints enable testing strategies. If we hadn't constrained the agent to a narrow operation subset (Part 1), regex conversion would be infeasible. Architectural decisions early in the pipeline create options later.
  1. Imperfect testing that runs in seconds beats perfect testing that takes minutes. The pandas conversion isn't semantically identical to PySpark. It doesn't need to be. It needs to catch the bugs that matter before expensive validation.
  1. Flag what you can't test, don't hide it. The  # UNCONVERTED  pattern makes gaps visible. Silent gaps are dangerous; explicit gaps are manageable.
  1. Feed failures back into generation. Testing isn't just a gate — it's a signal. When code fails, the reason it failed becomes input to the next generation attempt.
  1. Code Interpreter is your universal test sandbox. For any AI-generated Python code — or code whose logic can be validated in Python — AgentCore Code Interpreter provides sub-10-second isolated execution at negligible cost.
  1. Layer validation by cost. Cheapest first, most expensive last. Every layer is a gate that prevents wasted spend at the next layer.

Built with Amazon Bedrock, AgentCore Code Interpreter, Strands Agents, and AWS Step Functions. The testing patterns described here run as Step 6 of a 10-step automated pipeline that generates, validates, and deploys ETL jobs without human code authorship.
Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.
Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article