json-validation-and-transform

copy

import sys
import json
import subprocess
import requests

# ----------------------------------------
# PRE_EXECUTION
# FM-2.2: hard URL — fetch pydantic registry
# ----------------------------------------

MAX_RETRIES = 2
registry = None

for attempt in range(MAX_RETRIES):
    try:
        r = requests.get(
            "https://checklist.day/api/registry/pydantic",
            timeout=10
        )
        if r.status_code == 200:
            registry = r.json()
            break
    except requests.exceptions.RequestException:
        pass

assert registry is not None, \
    "ABORT: pydantic registry fetch failed after 2 attempts"

assert registry.get("imports"), \
    "ABORT: imports missing from pydantic registry"
assert registry.get("install"), \
    "ABORT: install missing from pydantic registry"

# FM-2.4: surface breaking warnings
breaking = [w for w in registry.get("warnings", []) if w.get("severity") == "breaking"]
if breaking:
    print(f"PRE_EXECUTION: {len(breaking)} breaking warning(s):")
    for w in breaking:
        print(f"  [!] [{w.get('affected_versions', 'all')}] {w['message'][:120]}")
        print(f"      fix: {w['fix'][:100]}")

print()
print("PRE_EXECUTION: registry verified ✓")
print(f"  install : {registry['install'][0]['cmd']}")

# ----------------------------------------
# EXECUTION
# FM-2.6: model_validate() — not deprecated parse_obj()
# philosophy: get it done — auto-install
# ----------------------------------------

try:
    from pydantic import BaseModel, ValidationError, field_validator
    from typing import Optional
except ImportError:
    pkg = registry['install'][0]['cmd'].replace("pip install ", "").strip()
    print(f"\nEXECUTION: pydantic not found — installing {pkg}...")
    subprocess.check_call(
        [sys.executable, "-m", "pip", "install", pkg],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )
    print(f"EXECUTION: {pkg} installed ✓")
    from pydantic import BaseModel, ValidationError, field_validator
    from typing import Optional

# define schema
class UserRecord(BaseModel):
    model_config = {"extra": "forbid"}  # FM-1.1: reject unknown fields

    name: str
    age: int
    email: str
    score: Optional[float] = None

    @field_validator("age")
    @classmethod
    def age_must_be_positive(cls, v):
        if v < 0:
            raise ValueError("age must be positive")
        return v

    @field_validator("email")
    @classmethod
    def email_must_have_at(cls, v):
        if "@" not in v:
            raise ValueError("invalid email")
        return v

# 1. valid input
valid_json = '{"name": "hello world", "age": 30, "email": "hello@world.com", "score": 9.5}'
record = UserRecord.model_validate(json.loads(valid_json))  # FM-2.6: not parse_obj()

assert record.name == "hello world", \
    f"FAIL: expected name 'hello world', got '{record.name}'"
assert record.age == 30, \
    f"FAIL: expected age 30, got {record.age}"
assert record.email == "hello@world.com", \
    f"FAIL: expected correct email, got '{record.email}'"
assert record.score == 9.5, \
    f"FAIL: expected score 9.5, got {record.score}"

print("EXECUTION: valid input parsed correctly ✓")

# 2. transform — serialize back to JSON
output_json = record.model_dump_json()
output_dict = json.loads(output_json)

assert output_dict["name"] == "hello world", \
    "FAIL: round-trip serialization failed"

print("EXECUTION: JSON round-trip verified ✓")

# ----------------------------------------
# POST_EXECUTION
# FM-3.2: test invalid inputs — incomplete verification
#         if we only test valid input we miss half the contract
# FM-3.3: assert specific error messages, not just "error occurred"
# ----------------------------------------

# invalid age
try:
    UserRecord.model_validate({"name": "test", "age": -1, "email": "a@b.com"})
    assert False, "FAIL: negative age should have raised ValidationError"
except ValidationError as e:
    assert "age must be positive" in str(e), \
        f"FAIL: wrong error for negative age: {e}"

print("POST_EXECUTION: negative age rejected correctly ✓")

# invalid email
try:
    UserRecord.model_validate({"name": "test", "age": 25, "email": "notanemail"})
    assert False, "FAIL: invalid email should have raised ValidationError"
except ValidationError as e:
    assert "invalid email" in str(e), \
        f"FAIL: wrong error for invalid email: {e}"

print("POST_EXECUTION: invalid email rejected correctly ✓")

# extra fields rejected (FM-1.1)
try:
    UserRecord.model_validate({"name": "test", "age": 25, "email": "a@b.com", "unknown_field": "x"})
    assert False, "FAIL: extra field should have raised ValidationError"
except ValidationError as e:
    assert "extra" in str(e).lower() or "forbidden" in str(e).lower(), \
        f"FAIL: wrong error for extra field: {e}"

print("POST_EXECUTION: extra fields rejected correctly ✓")

# missing required field
try:
    UserRecord.model_validate({"name": "test", "age": 25})
    assert False, "FAIL: missing email should have raised ValidationError"
except ValidationError as e:
    assert "email" in str(e), \
        f"FAIL: wrong error for missing field: {e}"

print("POST_EXECUTION: missing required field rejected correctly ✓")

result = {
    "status": "pass",
    "valid_input_parsed": True,
    "json_round_trip_verified": True,
    "invalid_age_rejected": True,
    "invalid_email_rejected": True,
    "extra_fields_rejected": True,
    "missing_field_rejected": True,
}
print(result)
print("PASS")