Test Generation Pipeline

# file: tests/test_structural.py
#
# AUTO-GENERATED from spec: UrlShortener
# Do not edit manually -- regenerate with: spec-to-rest generate --tests

import schemathesis
import pytest
from urllib.parse import urlparse

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

BASE_URL = "http://localhost:8000"
SCHEMA_PATH = "./openapi.json"

schema = schemathesis.from_path(
    SCHEMA_PATH,
    base_url=BASE_URL,
)

# ---------------------------------------------------------------------------
# Test profile (set via environment variable SPEC_TEST_PROFILE)
# ---------------------------------------------------------------------------

import os

PROFILE = os.environ.get("SPEC_TEST_PROFILE", "thorough")

PROFILE_SETTINGS = {
    "smoke": {"max_examples": 10, "stateful_step_count": 3},
    "thorough": {"max_examples": 100, "stateful_step_count": 10},
    "exhaustive": {"max_examples": 1000, "stateful_step_count": 25},
}

settings = PROFILE_SETTINGS[PROFILE]

# ---------------------------------------------------------------------------
# Auth hook (if the spec declares authentication)
# ---------------------------------------------------------------------------

AUTH_TOKEN = os.environ.get("SPEC_TEST_AUTH_TOKEN", None)


@schemathesis.hook("before_call")
def add_auth(context, case, **kwargs):
    """Attach auth token to every request if configured."""
    if AUTH_TOKEN is not None:
        case.headers = case.headers or {}
        case.headers["Authorization"] = f"Bearer {AUTH_TOKEN}"


# ---------------------------------------------------------------------------
# Custom checks derived from spec ensures/invariants
# ---------------------------------------------------------------------------


@schemathesis.check
def spec_shorten_returns_valid_short_url(response, case):
    """
    Derived from: ensures short_url = base_url + "/" + code.value
    Only applies to POST /shorten with 201 responses.
    """
    if case.path != "/shorten" or case.method.upper() != "POST":
        return
    if response.status_code != 201:
        return
    data = response.json()
    short_url = data.get("short_url", "")
    code = data.get("code", "")
    assert short_url.endswith(
        f"/{code}"
    ), f"short_url '{short_url}' does not end with code '{code}'"
    parsed = urlparse(short_url)
    assert parsed.scheme in (
        "http",
        "https",
    ), f"short_url '{short_url}' has invalid scheme"


@schemathesis.check
def spec_shorten_code_length_valid(response, case):
    """
    Derived from entity invariant: len(ShortCode.value) >= 6 and len(ShortCode.value) <= 10
    """
    if case.path != "/shorten" or case.method.upper() != "POST":
        return
    if response.status_code != 201:
        return
    code = response.json().get("code", "")
    assert 6 <= len(code) <= 10, f"code length {len(code)} outside [6, 10]"


@schemathesis.check
def spec_shorten_code_alphanumeric(response, case):
    """
    Derived from entity invariant: ShortCode.value matches /^[a-zA-Z0-9]+$/
    """
    if case.path != "/shorten" or case.method.upper() != "POST":
        return
    if response.status_code != 201:
        return
    code = response.json().get("code", "")
    assert code.isalnum(), f"code '{code}' is not alphanumeric"


@schemathesis.check
def spec_resolve_returns_valid_uri(response, case):
    """
    Derived from: invariant all c in store | isValidURI(store[c].value)
    Applies to GET /{code} with 302 responses.
    """
    if not case.path.startswith("/{code}"):
        return
    if case.method.upper() != "GET":
        return
    if response.status_code != 302:
        return
    location = response.headers.get("Location", "")
    parsed = urlparse(location)
    assert parsed.scheme in (
        "http",
        "https",
    ), f"Location header '{location}' is not a valid URI"


@schemathesis.check
def spec_no_server_errors(response, case):
    """All operations: server must never return 5xx."""
    assert response.status_code < 500, (
        f"Server error {response.status_code} on "
        f"{case.method.upper()} {case.path}"
    )


# ---------------------------------------------------------------------------
# Stateless tests (one per endpoint)
# ---------------------------------------------------------------------------


from hypothesis import settings as hypothesis_settings

@schema.parametrize()
@hypothesis_settings(max_examples=settings["max_examples"])
def test_api_structural(case):
    """
    For each endpoint in the OpenAPI spec, Schemathesis generates random
    valid requests and validates that responses conform to the declared schemas.
    Custom checks above are run automatically for every response.
    """
    case.call_and_validate()


# ---------------------------------------------------------------------------
# Stateful tests (multi-step sequences via OpenAPI links)
# ---------------------------------------------------------------------------
#
# The generated OpenAPI spec includes links that express the data flow
# between operations:
#
#   POST /shorten  --[code]--> GET /{code}
#   POST /shorten  --[code]--> DELETE /{code}
#   GET /{code}    --[code]--> DELETE /{code}
#
# Schemathesis discovers these links and generates multi-step test sequences.

state_machine = schema.as_state_machine()


class TestStatefulUrlShortener(state_machine):
    """
    Stateful test class. Schemathesis builds a state machine from the OpenAPI
    links and explores multi-step sequences up to the configured depth.
    """

    pass


TestStatefulUrlShortener.settings = hypothesis_settings(
    max_examples=settings["max_examples"],
    stateful_step_count=settings["stateful_step_count"],
)

# ---------------------------------------------------------------------------
# Database seeding fixture
# ---------------------------------------------------------------------------


@pytest.fixture(scope="session", autouse=True)
def seed_database():
    """
    Seed the database with known test data before the test session.
    Generated from the spec's initial state constraints.
    """
    import httpx

    # Seed a few known URLs so that resolve/delete tests have targets
    seed_data = [
        {"url": "https://example.com/page1"},
        {"url": "https://example.com/page2"},
        {"url": "https://example.com/page3"},
    ]
    created_codes = []
    for item in seed_data:
        resp = httpx.post(f"{BASE_URL}/shorten", json=item)
        if resp.status_code == 201:
            created_codes.append(resp.json()["code"])

    yield created_codes

    # Teardown: delete seeded entries
    for code in created_codes:
        httpx.delete(f"{BASE_URL}/{code}")

.PHONY: test-structural test-structural-smoke test-structural-exhaustive

test-structural-smoke:
	SPEC_TEST_PROFILE=smoke pytest tests/test_structural.py -v --tb=short

test-structural:
	SPEC_TEST_PROFILE=thorough pytest tests/test_structural.py -v --tb=short

test-structural-exhaustive:
	SPEC_TEST_PROFILE=exhaustive pytest tests/test_structural.py -v --tb=long

# file: tests/conftest.py
#
# AUTO-GENERATED from spec: UrlShortener

import pytest
import httpx
from hypothesis import strategies as st, given, settings, assume, HealthCheck

# ---------------------------------------------------------------------------
# Test client
# ---------------------------------------------------------------------------

BASE_URL = "http://localhost:8000"


@pytest.fixture(scope="session")
def client():
    with httpx.Client(base_url=BASE_URL, timeout=10.0) as c:
        yield c


# Global client for use in Hypothesis tests (which cannot use pytest fixtures
# directly due to the @given decorator)
_client = httpx.Client(base_url=BASE_URL, timeout=10.0)


def get_client() -> httpx.Client:
    return _client


# ---------------------------------------------------------------------------
# Spec-derived helper functions
# ---------------------------------------------------------------------------


def get_all_codes() -> set[str]:
    """Return all short codes currently in the store."""
    resp = get_client().get("/codes")
    resp.raise_for_status()
    return set(resp.json())


def get_store_count() -> int:
    """Return the number of entries in the store."""
    resp = get_client().get("/stats")
    resp.raise_for_status()
    return resp.json()["count"]


def is_valid_uri(s: str) -> bool:
    """Check if a string is a valid URI per the spec's valid_uri predicate."""
    from urllib.parse import urlparse

    try:
        result = urlparse(s)
        return all([result.scheme in ("http", "https"), result.netloc])
    except Exception:
        return False


# ---------------------------------------------------------------------------
# Database reset between tests
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def reset_state():
    """Reset the service state before each test function."""
    resp = get_client().post("/test/reset")
    assert resp.status_code == 200
    yield
    # No teardown needed; next test will reset

service UrlShortener {
  entity ShortCode { value: String; invariant: len(value) >= 6 and len(value) <= 10; invariant: value matches /^[a-zA-Z0-9]+$/ }
  entity LongURL   { value: String; invariant: isValidURI(value) }
  state { store: ShortCode -> lone LongURL }

  operation Shorten {
    input:  url: LongURL
    output: code: ShortCode, short_url: String
    requires: isValidURI(url.value)
    ensures:  code not in pre(store); store'[code] = url; #store' = #store + 1
  }

  operation Resolve {
    input:  code: ShortCode
    output: url: LongURL
    requires: code in store
    ensures:  url = store[code]; store' = store
  }

  operation Delete {
    input:  code: ShortCode
    requires: code in store
    ensures:  code not in store'; #store' = #store - 1
  }

  invariant: all c in store | isValidURI(store[c].value)
}

# file: tests/test_behavioral_url_shortener.py
#
# AUTO-GENERATED from spec: UrlShortener
# Tests every ensures clause as a Hypothesis property test,
# every requires clause as a negative test,
# and every invariant as a post-operation check.

import re
import pytest
from hypothesis import given, settings, assume, HealthCheck
from hypothesis import strategies as st
from tests.conftest import get_client, get_all_codes, get_store_count, is_valid_uri

# ---------------------------------------------------------------------------
# Strategies derived from entity definitions
# ---------------------------------------------------------------------------

valid_urls = st.sampled_from([
    "https://example.com",
    "https://example.com/path/to/page",
    "https://subdomain.example.org/page?q=1&r=2",
    "http://example.net/a/b/c",
    "https://example.com/very/long/path/" + "x" * 200,
]) | st.from_regex(
    r"https?://[a-z]{3,12}\.[a-z]{2,6}(/[a-z0-9]{1,20}){0,5}",
    fullmatch=True,
)

invalid_urls = st.sampled_from([
    "",
    "not-a-url",
    "ftp://wrong-scheme.com",
    "://missing-scheme",
    "http://",
    "   ",
    "javascript:alert(1)",
])

# ---------------------------------------------------------------------------
# OPERATION: Shorten
# ---------------------------------------------------------------------------
# ensures: code not in pre(store)
# ensures: store'[code] = url
# ensures: #store' = #store + 1


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_shorten_returns_fresh_code(url):
    """
    Ensures: code not in pre(store)

    After shortening, the returned code must not have existed before
    the operation was called.
    """
    client = get_client()
    client.post("/test/reset")

    # Capture pre-state
    existing_codes = get_all_codes()

    # Execute operation
    response = client.post("/shorten", json={"url": url})
    assert response.status_code == 201

    code = response.json()["code"]
    assert code not in existing_codes, (
        f"Returned code '{code}' already existed in store before shorten was called"
    )


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_shorten_stores_correct_mapping(url):
    """
    Ensures: store'[code] = url

    After shortening, resolving the returned code must yield the original URL.
    """
    client = get_client()
    client.post("/test/reset")

    response = client.post("/shorten", json={"url": url})
    assert response.status_code == 201

    code = response.json()["code"]
    lookup = client.get(f"/{code}", follow_redirects=False)
    assert lookup.status_code == 302
    assert lookup.headers["Location"] == url, (
        f"Expected Location '{url}', got '{lookup.headers.get('Location')}'"
    )


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_shorten_adds_exactly_one(url):
    """
    Ensures: #store' = #store + 1

    The store must contain exactly one more entry after a successful shorten.
    """
    client = get_client()
    client.post("/test/reset")

    count_before = get_store_count()
    response = client.post("/shorten", json={"url": url})
    assert response.status_code == 201

    count_after = get_store_count()
    assert count_after == count_before + 1, (
        f"Store count went from {count_before} to {count_after}, expected {count_before + 1}"
    )


# ---------------------------------------------------------------------------
# OPERATION: Resolve
# ---------------------------------------------------------------------------
# ensures: url = store[code]
# ensures: store' = store


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_resolve_returns_correct_url(url):
    """
    Ensures: url = store[code]

    Resolving a code must return the URL that was stored when it was shortened.
    """
    client = get_client()
    client.post("/test/reset")

    # Create a mapping
    shorten_resp = client.post("/shorten", json={"url": url})
    assert shorten_resp.status_code == 201
    code = shorten_resp.json()["code"]

    # Resolve it
    resolve_resp = client.get(f"/{code}", follow_redirects=False)
    assert resolve_resp.status_code == 302
    assert resolve_resp.headers["Location"] == url


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_resolve_does_not_mutate_state(url):
    """
    Ensures: store' = store

    The resolve operation must not change the store. The count before and after
    must be identical.
    """
    client = get_client()
    client.post("/test/reset")

    # Create a mapping
    shorten_resp = client.post("/shorten", json={"url": url})
    assert shorten_resp.status_code == 201
    code = shorten_resp.json()["code"]

    count_before = get_store_count()
    codes_before = get_all_codes()

    # Resolve
    client.get(f"/{code}", follow_redirects=False)

    count_after = get_store_count()
    codes_after = get_all_codes()

    assert count_after == count_before, "Resolve mutated the store count"
    assert codes_after == codes_before, "Resolve mutated the store contents"


# ---------------------------------------------------------------------------
# OPERATION: Delete
# ---------------------------------------------------------------------------
# ensures: code not in store'
# ensures: #store' = #store - 1


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_delete_removes_code(url):
    """
    Ensures: code not in store'

    After deletion, the code must no longer be in the store.
    """
    client = get_client()
    client.post("/test/reset")

    # Create a mapping
    shorten_resp = client.post("/shorten", json={"url": url})
    assert shorten_resp.status_code == 201
    code = shorten_resp.json()["code"]

    # Delete it
    delete_resp = client.delete(f"/{code}")
    assert delete_resp.status_code == 204

    # Verify removal
    remaining_codes = get_all_codes()
    assert code not in remaining_codes, f"Code '{code}' still in store after delete"


@given(url=valid_urls)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_delete_removes_exactly_one(url):
    """
    Ensures: #store' = #store - 1

    The store must contain exactly one fewer entry after a successful delete.
    """
    client = get_client()
    client.post("/test/reset")

    # Create a mapping
    shorten_resp = client.post("/shorten", json={"url": url})
    assert shorten_resp.status_code == 201
    code = shorten_resp.json()["code"]

    count_before = get_store_count()
    delete_resp = client.delete(f"/{code}")
    assert delete_resp.status_code == 204

    count_after = get_store_count()
    assert count_after == count_before - 1, (
        f"Store count went from {count_before} to {count_after}, expected {count_before - 1}"
    )


# ---------------------------------------------------------------------------
# NEGATIVE TESTS: requires clause violations
# ---------------------------------------------------------------------------


@given(url=invalid_urls)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_shorten_rejects_invalid_url(url):
    """
    Requires: isValidURI(url.value)

    Violating this precondition must return 422 and must NOT mutate state.
    """
    client = get_client()
    client.post("/test/reset")

    count_before = get_store_count()
    response = client.post("/shorten", json={"url": url})
    assert response.status_code == 422, (
        f"Expected 422 for invalid URL '{url}', got {response.status_code}"
    )

    count_after = get_store_count()
    assert count_after == count_before, (
        "State was mutated despite requires clause violation"
    )


def test_resolve_rejects_nonexistent_code():
    """
    Requires: code in store

    Resolving a code that does not exist must return 404.
    """
    client = get_client()
    client.post("/test/reset")

    response = client.get("/NONEXISTENT", follow_redirects=False)
    assert response.status_code == 404


def test_delete_rejects_nonexistent_code():
    """
    Requires: code in store

    Deleting a code that does not exist must return 404 and must NOT mutate state.
    """
    client = get_client()
    client.post("/test/reset")

    count_before = get_store_count()
    response = client.delete("/NONEXISTENT")
    assert response.status_code == 404

    count_after = get_store_count()
    assert count_after == count_before, (
        "State was mutated despite requires clause violation"
    )


# ---------------------------------------------------------------------------
# INVARIANT CHECKS: run after every operation
# ---------------------------------------------------------------------------


def check_global_invariants():
    """
    Invariant: all c in store | isValidURI(store[c].value)

    Every URL in the store must be a valid URI. Called after every operation
    in the stateful test layer, and spot-checked here.
    """
    client = get_client()
    codes = get_all_codes()
    for code in codes:
        resp = client.get(f"/{code}", follow_redirects=False)
        if resp.status_code == 302:
            location = resp.headers["Location"]
            assert is_valid_uri(location), (
                f"Invariant violated: store[{code}] = '{location}' is not a valid URI"
            )


@given(url=valid_urls)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_invariant_holds_after_shorten(url):
    """Global invariant must hold after every Shorten operation."""
    client = get_client()
    client.post("/test/reset")
    client.post("/shorten", json={"url": url})
    check_global_invariants()


@given(url=valid_urls)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_invariant_holds_after_delete(url):
    """Global invariant must hold after every Delete operation."""
    client = get_client()
    client.post("/test/reset")

    resp = client.post("/shorten", json={"url": url})
    assert resp.status_code == 201
    code = resp.json()["code"]
    client.delete(f"/{code}")
    check_global_invariants()

service TodoList {
  entity TodoId    { value: UUID }
  entity TodoTitle { value: String; invariant: len(value) >= 1 and len(value) <= 200 }

  enum TodoStatus { todo, in_progress, done }

  entity TodoItem {
    id:     TodoId
    title:  TodoTitle
    status: TodoStatus
  }

  state { items: TodoId -> lone TodoItem }

  operation CreateTodo {
    input:  title: TodoTitle
    output: item: TodoItem
    requires: len(title.value) >= 1
    ensures:
      item.id not in pre(items)
      item.status = TodoStatus.todo
      item.title = title
      items'[item.id] = item
      #items' = #items + 1
  }

  operation StartTodo {
    input:  id: TodoId
    output: item: TodoItem
    requires: id in items
    requires: items[id].status = TodoStatus.todo
    ensures:
      item.status = TodoStatus.in_progress
      items'[id].status = TodoStatus.in_progress
      #items' = #items
  }

  operation CompleteTodo {
    input:  id: TodoId
    output: item: TodoItem
    requires: id in items
    requires: items[id].status = TodoStatus.in_progress
    ensures:
      item.status = TodoStatus.done
      items'[id].status = TodoStatus.done
      #items' = #items
  }

  operation DeleteTodo {
    input:  id: TodoId
    requires: id in items
    ensures:
      id not in items'
      #items' = #items - 1
  }

  // State transition constraints
  invariant: all t in items | items[t].status = done implies
    (items[t] was in_progress)  // done items must have been in_progress

  // General invariants
  invariant: all t in items | len(items[t].title.value) >= 1
}

# file: tests/test_behavioral_todo.py
#
# AUTO-GENERATED from spec: TodoList

import pytest
from hypothesis import given, settings, assume, HealthCheck
from hypothesis import strategies as st
from tests.conftest import get_client

# ---------------------------------------------------------------------------
# Strategies
# ---------------------------------------------------------------------------

valid_titles = st.text(
    alphabet=st.characters(whitelist_categories=("L", "N", "P", "Z")),
    min_size=1,
    max_size=200,
).filter(lambda t: len(t.strip()) >= 1)

invalid_titles = st.sampled_from(["", None]) | st.text(min_size=201, max_size=300)

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def get_all_items():
    resp = get_client().get("/todos")
    resp.raise_for_status()
    return resp.json()


def get_item_count():
    return len(get_all_items())


# ---------------------------------------------------------------------------
# OPERATION: CreateTodo
# ---------------------------------------------------------------------------
# ensures: item.id not in pre(items)
# ensures: item.status = TodoStatus.todo
# ensures: item.title = title
# ensures: items'[item.id] = item
# ensures: #items' = #items + 1


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_create_todo_returns_fresh_id(title):
    """Ensures: item.id not in pre(items)"""
    client = get_client()
    client.post("/test/reset")

    existing_ids = {item["id"] for item in get_all_items()}
    response = client.post("/todos", json={"title": title})
    assert response.status_code == 201

    new_id = response.json()["id"]
    assert new_id not in existing_ids, f"ID '{new_id}' already existed"


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_create_todo_initial_status_is_todo(title):
    """Ensures: item.status = TodoStatus.todo"""
    client = get_client()
    client.post("/test/reset")

    response = client.post("/todos", json={"title": title})
    assert response.status_code == 201

    item = response.json()
    assert item["status"] == "todo", f"Expected status 'todo', got '{item['status']}'"


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_create_todo_stores_correct_title(title):
    """Ensures: item.title = title"""
    client = get_client()
    client.post("/test/reset")

    response = client.post("/todos", json={"title": title})
    assert response.status_code == 201

    item = response.json()
    assert item["title"] == title, f"Expected title '{title}', got '{item['title']}'"


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_create_todo_adds_exactly_one(title):
    """Ensures: #items' = #items + 1"""
    client = get_client()
    client.post("/test/reset")

    count_before = get_item_count()
    response = client.post("/todos", json={"title": title})
    assert response.status_code == 201

    count_after = get_item_count()
    assert count_after == count_before + 1


# ---------------------------------------------------------------------------
# OPERATION: StartTodo
# ---------------------------------------------------------------------------
# ensures: item.status = TodoStatus.in_progress
# ensures: items'[id].status = TodoStatus.in_progress
# ensures: #items' = #items


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_start_todo_transitions_to_in_progress(title):
    """Ensures: item.status = TodoStatus.in_progress"""
    client = get_client()
    client.post("/test/reset")

    # Create a todo (starts in 'todo' state)
    create_resp = client.post("/todos", json={"title": title})
    assert create_resp.status_code == 201
    todo_id = create_resp.json()["id"]

    # Start it
    start_resp = client.patch(f"/todos/{todo_id}/start")
    assert start_resp.status_code == 200

    item = start_resp.json()
    assert item["status"] == "in_progress"


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_start_todo_persists_status(title):
    """Ensures: items'[id].status = TodoStatus.in_progress (check via GET)"""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]
    client.patch(f"/todos/{todo_id}/start")

    # Verify persisted state via a fresh GET
    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.status_code == 200
    assert get_resp.json()["status"] == "in_progress"


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_start_todo_does_not_change_count(title):
    """Ensures: #items' = #items"""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]

    count_before = get_item_count()
    client.patch(f"/todos/{todo_id}/start")
    count_after = get_item_count()

    assert count_after == count_before


# ---------------------------------------------------------------------------
# OPERATION: CompleteTodo
# ---------------------------------------------------------------------------
# ensures: item.status = TodoStatus.done
# ensures: items'[id].status = TodoStatus.done
# ensures: #items' = #items


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_complete_todo_transitions_to_done(title):
    """Ensures: item.status = TodoStatus.done"""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]
    client.patch(f"/todos/{todo_id}/start")  # must be in_progress first

    complete_resp = client.patch(f"/todos/{todo_id}/complete")
    assert complete_resp.status_code == 200
    assert complete_resp.json()["status"] == "done"


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_complete_todo_persists_status(title):
    """Ensures: items'[id].status = TodoStatus.done (check via GET)"""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]
    client.patch(f"/todos/{todo_id}/start")
    client.patch(f"/todos/{todo_id}/complete")

    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.json()["status"] == "done"


# ---------------------------------------------------------------------------
# OPERATION: DeleteTodo
# ---------------------------------------------------------------------------
# ensures: id not in items'
# ensures: #items' = #items - 1


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_delete_todo_removes_item(title):
    """Ensures: id not in items'"""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]

    delete_resp = client.delete(f"/todos/{todo_id}")
    assert delete_resp.status_code == 204

    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.status_code == 404


@given(title=valid_titles)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_delete_todo_removes_exactly_one(title):
    """Ensures: #items' = #items - 1"""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]

    count_before = get_item_count()
    client.delete(f"/todos/{todo_id}")
    count_after = get_item_count()

    assert count_after == count_before - 1


# ---------------------------------------------------------------------------
# NEGATIVE TESTS: requires clause violations
# ---------------------------------------------------------------------------


def test_start_rejects_nonexistent_todo():
    """Requires: id in items"""
    client = get_client()
    client.post("/test/reset")
    resp = client.patch("/todos/00000000-0000-0000-0000-000000000000/start")
    assert resp.status_code == 404


@given(title=valid_titles)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_start_rejects_non_todo_status(title):
    """Requires: items[id].status = TodoStatus.todo -- cannot start an in_progress item."""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]
    client.patch(f"/todos/{todo_id}/start")  # now in_progress

    # Trying to start again should fail
    resp = client.patch(f"/todos/{todo_id}/start")
    assert resp.status_code == 422 or resp.status_code == 409, (
        f"Expected 422 or 409 for invalid transition, got {resp.status_code}"
    )


@given(title=valid_titles)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_complete_rejects_todo_status(title):
    """Requires: items[id].status = TodoStatus.in_progress -- cannot complete from todo."""
    client = get_client()
    client.post("/test/reset")

    create_resp = client.post("/todos", json={"title": title})
    todo_id = create_resp.json()["id"]

    # Item is in 'todo' status, not 'in_progress'
    resp = client.patch(f"/todos/{todo_id}/complete")
    assert resp.status_code == 422 or resp.status_code == 409

    # Verify state was not mutated
    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.json()["status"] == "todo"


@given(title=invalid_titles)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_create_rejects_invalid_title(title):
    """Requires: len(title.value) >= 1"""
    client = get_client()
    client.post("/test/reset")

    count_before = get_item_count()
    resp = client.post("/todos", json={"title": title})
    assert resp.status_code == 422

    count_after = get_item_count()
    assert count_after == count_before, "State mutated despite requires failure"

service OrderSystem {
  entity ProductId   { value: UUID }
  entity OrderId     { value: UUID }
  entity CustomerId  { value: UUID }
  entity Money       { cents: Int; invariant: cents >= 0 }

  enum OrderStatus { pending, confirmed, shipped, delivered, cancelled }

  entity OrderLine {
    product: ProductId
    quantity: Int
    unit_price: Money
    invariant: quantity >= 1
  }

  entity Order {
    id:       OrderId
    customer: CustomerId
    lines:    set OrderLine
    status:   OrderStatus
    total:    Money
  }

  state {
    orders:   OrderId -> lone Order
    products: ProductId -> lone Product    // catalog (read-only for order ops)
    inventory: ProductId -> Int            // stock counts
  }

  operation PlaceOrder {
    input:  customer: CustomerId, lines: set OrderLine
    output: order: Order
    requires: #lines >= 1
    requires: all l in lines | l.product in products
    requires: all l in lines | inventory[l.product] >= l.quantity
    ensures:
      order.id not in pre(orders)
      order.status = OrderStatus.pending
      order.customer = customer
      order.lines = lines
      order.total.cents = sum(l in lines | l.quantity * l.unit_price.cents)
      orders'[order.id] = order
      all l in lines | inventory'[l.product] = inventory[l.product] - l.quantity
      #orders' = #orders + 1
  }

  operation ConfirmOrder {
    input:  id: OrderId
    output: order: Order
    requires: id in orders
    requires: orders[id].status = OrderStatus.pending
    ensures:
      order.status = OrderStatus.confirmed
      orders'[id].status = OrderStatus.confirmed
  }

  operation CancelOrder {
    input:  id: OrderId
    output: order: Order
    requires: id in orders
    requires: orders[id].status in {OrderStatus.pending, OrderStatus.confirmed}
    ensures:
      order.status = OrderStatus.cancelled
      orders'[id].status = OrderStatus.cancelled
      // Restore inventory
      all l in orders[id].lines | inventory'[l.product] = inventory[l.product] + l.quantity
  }

  invariant: all o in orders | orders[o].total.cents >= 0
  invariant: all o in orders | #orders[o].lines >= 1
  invariant: all p in inventory | inventory[p] >= 0
}

# file: tests/test_behavioral_orders.py
#
# AUTO-GENERATED from spec: OrderSystem

import pytest
from hypothesis import given, settings, assume, HealthCheck
from hypothesis import strategies as st
from tests.conftest import get_client

# ---------------------------------------------------------------------------
# Strategies
# ---------------------------------------------------------------------------

# These assume a pre-seeded product catalog with known product IDs and prices.
SEED_PRODUCTS = [
    {"id": "prod-001", "name": "Widget", "price_cents": 999},
    {"id": "prod-002", "name": "Gadget", "price_cents": 2499},
    {"id": "prod-003", "name": "Doohickey", "price_cents": 50},
]

CUSTOMER_ID = "cust-001"

order_lines = st.lists(
    st.fixed_dictionaries({
        "product": st.sampled_from([p["id"] for p in SEED_PRODUCTS]),
        "quantity": st.integers(min_value=1, max_value=5),
        "unit_price_cents": st.sampled_from([p["price_cents"] for p in SEED_PRODUCTS]),
    }),
    min_size=1,
    max_size=5,
)

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def get_all_orders():
    resp = get_client().get("/orders")
    resp.raise_for_status()
    return resp.json()


def get_order_count():
    return len(get_all_orders())


def get_inventory(product_id: str) -> int:
    resp = get_client().get(f"/products/{product_id}/inventory")
    resp.raise_for_status()
    return resp.json()["quantity"]


def seed_catalog():
    """Ensure seed products exist with sufficient inventory."""
    client = get_client()
    for p in SEED_PRODUCTS:
        client.put(f"/products/{p['id']}", json={
            "name": p["name"],
            "price_cents": p["price_cents"],
        })
        client.put(f"/products/{p['id']}/inventory", json={"quantity": 1000})


# ---------------------------------------------------------------------------
# OPERATION: PlaceOrder
# ---------------------------------------------------------------------------
# ensures: order.id not in pre(orders)
# ensures: order.status = OrderStatus.pending
# ensures: order.customer = customer
# ensures: order.total.cents = sum(...)
# ensures: orders'[order.id] = order
# ensures: all l in lines | inventory'[l.product] = inventory[l.product] - l.quantity
# ensures: #orders' = #orders + 1


@given(lines=order_lines)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_place_order_returns_fresh_id(lines):
    """Ensures: order.id not in pre(orders)"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    existing_ids = {o["id"] for o in get_all_orders()}
    resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    assert resp.status_code == 201

    new_id = resp.json()["id"]
    assert new_id not in existing_ids


@given(lines=order_lines)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_place_order_initial_status_pending(lines):
    """Ensures: order.status = OrderStatus.pending"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    assert resp.status_code == 201
    assert resp.json()["status"] == "pending"


@given(lines=order_lines)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_place_order_correct_total(lines):
    """Ensures: order.total.cents = sum(l in lines | l.quantity * l.unit_price.cents)"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    assert resp.status_code == 201

    order = resp.json()
    expected_total = sum(l["quantity"] * l["unit_price_cents"] for l in lines)
    assert order["total_cents"] == expected_total, (
        f"Expected total {expected_total}, got {order['total_cents']}"
    )


@given(lines=order_lines)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_place_order_decrements_inventory(lines):
    """Ensures: all l in lines | inventory'[l.product] = inventory[l.product] - l.quantity"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    # Record pre-state inventory
    inventory_before = {}
    for line in lines:
        pid = line["product"]
        if pid not in inventory_before:
            inventory_before[pid] = get_inventory(pid)

    resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    assert resp.status_code == 201

    # Check inventory decremented correctly
    expected_decrements = {}
    for line in lines:
        pid = line["product"]
        expected_decrements[pid] = expected_decrements.get(pid, 0) + line["quantity"]

    for pid, decrement in expected_decrements.items():
        inventory_after = get_inventory(pid)
        assert inventory_after == inventory_before[pid] - decrement, (
            f"Product {pid}: expected inventory {inventory_before[pid] - decrement}, "
            f"got {inventory_after}"
        )


@given(lines=order_lines)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_place_order_adds_exactly_one(lines):
    """Ensures: #orders' = #orders + 1"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    count_before = get_order_count()
    resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    assert resp.status_code == 201

    count_after = get_order_count()
    assert count_after == count_before + 1


# ---------------------------------------------------------------------------
# OPERATION: CancelOrder
# ---------------------------------------------------------------------------
# ensures: order.status = OrderStatus.cancelled
# ensures: all l in orders[id].lines | inventory'[l.product] = inventory[l.product] + l.quantity


@given(lines=order_lines)
@settings(
    max_examples=100,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_cancel_order_restores_inventory(lines):
    """
    Ensures: all l in orders[id].lines |
        inventory'[l.product] = inventory[l.product] + l.quantity

    Cancelling an order must restore the inventory that was decremented.
    """
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    # Record inventory before placing order
    inventory_initial = {}
    for line in lines:
        pid = line["product"]
        if pid not in inventory_initial:
            inventory_initial[pid] = get_inventory(pid)

    # Place and then cancel
    place_resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    assert place_resp.status_code == 201
    order_id = place_resp.json()["id"]

    cancel_resp = client.patch(f"/orders/{order_id}/cancel")
    assert cancel_resp.status_code == 200
    assert cancel_resp.json()["status"] == "cancelled"

    # Inventory should be fully restored
    for pid, initial_qty in inventory_initial.items():
        restored_qty = get_inventory(pid)
        assert restored_qty == initial_qty, (
            f"Product {pid}: inventory after cancel is {restored_qty}, "
            f"expected restoration to {initial_qty}"
        )


# ---------------------------------------------------------------------------
# NEGATIVE TESTS
# ---------------------------------------------------------------------------


def test_place_order_rejects_empty_lines():
    """Requires: #lines >= 1"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": []})
    assert resp.status_code == 422


def test_confirm_rejects_non_pending():
    """Requires: orders[id].status = OrderStatus.pending"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    # Create and confirm an order
    lines = [{"product": "prod-001", "quantity": 1, "unit_price_cents": 999}]
    place_resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    order_id = place_resp.json()["id"]
    client.patch(f"/orders/{order_id}/confirm")

    # Trying to confirm again should fail
    resp = client.patch(f"/orders/{order_id}/confirm")
    assert resp.status_code in (409, 422)


def test_cancel_rejects_shipped():
    """Requires: orders[id].status in {pending, confirmed}"""
    client = get_client()
    client.post("/test/reset")
    seed_catalog()

    lines = [{"product": "prod-001", "quantity": 1, "unit_price_cents": 999}]
    place_resp = client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    order_id = place_resp.json()["id"]
    client.patch(f"/orders/{order_id}/confirm")
    client.patch(f"/orders/{order_id}/ship")  # transition to shipped

    # Cannot cancel a shipped order
    resp = client.patch(f"/orders/{order_id}/cancel")
    assert resp.status_code in (409, 422)

    # Verify status was NOT changed
    get_resp = client.get(f"/orders/{order_id}")
    assert get_resp.json()["status"] == "shipped"


# ---------------------------------------------------------------------------
# INVARIANT CHECKS
# ---------------------------------------------------------------------------


def check_order_invariants():
    """
    invariant: all o in orders | orders[o].total.cents >= 0
    invariant: all o in orders | #orders[o].lines >= 1
    invariant: all p in inventory | inventory[p] >= 0
    """
    client = get_client()
    orders = get_all_orders()

    for order in orders:
        assert order["total_cents"] >= 0, (
            f"Order {order['id']} has negative total: {order['total_cents']}"
        )
        assert len(order["lines"]) >= 1, (
            f"Order {order['id']} has no lines"
        )

    for p in SEED_PRODUCTS:
        inv = get_inventory(p["id"])
        assert inv >= 0, f"Product {p['id']} has negative inventory: {inv}"


@given(lines=order_lines)
@settings(
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow, HealthCheck.function_scoped_fixture],
)
def test_invariants_hold_after_place_order(lines):
    client = get_client()
    client.post("/test/reset")
    seed_catalog()
    client.post("/orders", json={"customer": CUSTOMER_ID, "lines": lines})
    check_order_invariants()

# file: tests/test_stateful_url_shortener.py
#
# AUTO-GENERATED from spec: UrlShortener

import random
from hypothesis import settings, HealthCheck
from hypothesis import strategies as st
from hypothesis.stateful import (
    RuleBasedStateMachine,
    Bundle,
    rule,
    invariant,
    precondition,
    initialize,
)
from urllib.parse import urlparse
import httpx

BASE_URL = "http://localhost:8000"
client = httpx.Client(base_url=BASE_URL, timeout=10.0)

# Strategies derived from entity definitions
valid_urls = st.sampled_from([
    "https://example.com",
    "https://example.com/path/to/page",
    "https://subdomain.example.org/page?q=1&r=2",
    "http://example.net/a/b/c",
]) | st.from_regex(
    r"https?://[a-z]{3,12}\.[a-z]{2,6}(/[a-z0-9]{1,20}){0,5}",
    fullmatch=True,
)


def is_valid_uri(s: str) -> bool:
    try:
        result = urlparse(s)
        return all([result.scheme in ("http", "https"), result.netloc])
    except Exception:
        return False


class UrlShortenerStateMachine(RuleBasedStateMachine):
    """
    State machine that mirrors the spec's abstract state.
    The model state (self.store) tracks what the service should contain.
    Every rule executes a real HTTP request and verifies the response against
    both the model and the spec's ensures clauses.
    """

    # -----------------------------------------------------------------------
    # Model state (mirrors spec: state { store: ShortCode -> lone LongURL })
    # -----------------------------------------------------------------------
    def __init__(self):
        super().__init__()
        self.store: dict[str, str] = {}

    # -----------------------------------------------------------------------
    # Bundles for tracking created resources
    # -----------------------------------------------------------------------
    codes = Bundle("codes")

    # -----------------------------------------------------------------------
    # Initialize: reset the service state before each test case
    # -----------------------------------------------------------------------
    @initialize()
    def reset_service(self):
        resp = client.post("/test/reset")
        assert resp.status_code == 200
        self.store = {}

    # -----------------------------------------------------------------------
    # OPERATION: Shorten
    # Spec: requires isValidURI(url.value)
    #       ensures code not in pre(store)
    #       ensures store'[code] = url
    #       ensures #store' = #store + 1
    # -----------------------------------------------------------------------
    @rule(target=codes, url=valid_urls)
    def shorten(self, url):
        response = client.post("/shorten", json={"url": url})
        assert response.status_code == 201, (
            f"Shorten failed with {response.status_code}: {response.text}"
        )

        data = response.json()
        code = data["code"]

        # ensures: code not in pre(store)
        assert code not in self.store, (
            f"Code '{code}' already existed in model store"
        )

        # ensures: ShortCode entity invariants
        assert 6 <= len(code) <= 10, f"Code length {len(code)} outside [6, 10]"
        assert code.isalnum(), f"Code '{code}' is not alphanumeric"

        # ensures: short_url = base_url + "/" + code.value
        short_url = data.get("short_url", "")
        assert short_url.endswith(f"/{code}"), (
            f"short_url '{short_url}' does not end with code '{code}'"
        )

        # Update model state: store'[code] = url
        self.store[code] = url

        # ensures: #store' = #store + 1
        # (implicitly true since we added exactly one key that wasn't there)

        return code

    # -----------------------------------------------------------------------
    # OPERATION: Resolve
    # Spec: requires code in store
    #       ensures url = store[code]
    #       ensures store' = store
    # -----------------------------------------------------------------------
    @rule(code=codes)
    def resolve(self, code):
        # Only resolve codes that are still in the store
        if code not in self.store:
            # Code was deleted; expect 404
            response = client.get(f"/{code}", follow_redirects=False)
            assert response.status_code == 404, (
                f"Expected 404 for deleted code '{code}', got {response.status_code}"
            )
            return

        count_before = len(self.store)
        response = client.get(f"/{code}", follow_redirects=False)

        # ensures: url = store[code]
        assert response.status_code == 302, (
            f"Expected 302 for existing code '{code}', got {response.status_code}"
        )
        assert response.headers["Location"] == self.store[code], (
            f"Expected Location '{self.store[code]}', "
            f"got '{response.headers.get('Location')}'"
        )

        # ensures: store' = store (no mutation)
        assert len(self.store) == count_before

    # -----------------------------------------------------------------------
    # OPERATION: Delete
    # Spec: requires code in store
    #       ensures code not in store'
    #       ensures #store' = #store - 1
    # -----------------------------------------------------------------------
    @rule(code=codes)
    def delete(self, code):
        if code not in self.store:
            # Already deleted; expect 404
            response = client.delete(f"/{code}")
            assert response.status_code == 404
            return

        count_before = len(self.store)
        response = client.delete(f"/{code}")
        assert response.status_code == 204, (
            f"Expected 204 for delete of '{code}', got {response.status_code}"
        )

        # Update model: ensures code not in store'
        del self.store[code]

        # ensures: #store' = #store - 1
        assert len(self.store) == count_before - 1

    # -----------------------------------------------------------------------
    # NEGATIVE: Resolve nonexistent code
    # Tests: requires code in store -> 404 when violated
    # -----------------------------------------------------------------------
    @rule(code=st.text(
        alphabet=st.characters(whitelist_categories=("L", "N")),
        min_size=6,
        max_size=10,
    ))
    def resolve_nonexistent(self, code):
        if code in self.store:
            return  # Skip; this code happens to exist
        response = client.get(f"/{code}", follow_redirects=False)
        assert response.status_code == 404, (
            f"Expected 404 for nonexistent code '{code}', got {response.status_code}"
        )

    # -----------------------------------------------------------------------
    # INVARIANT: all c in store | isValidURI(store[c].value)
    # Checked after EVERY rule execution
    # -----------------------------------------------------------------------
    @invariant()
    def all_stored_urls_are_valid(self):
        for code, url in self.store.items():
            assert is_valid_uri(url), (
                f"Invariant violated: store[{code}] = '{url}' is not a valid URI"
            )

    # -----------------------------------------------------------------------
    # INVARIANT: model/service state consistency
    # The model store and the actual service store must agree
    # -----------------------------------------------------------------------
    @invariant()
    def model_matches_service(self):
        resp = client.get("/codes")
        if resp.status_code == 200:
            service_codes = set(resp.json())
            model_codes = set(self.store.keys())
            assert service_codes == model_codes, (
                f"Model/service mismatch. "
                f"Model has {model_codes - service_codes} extra, "
                f"service has {service_codes - model_codes} extra"
            )


# ---------------------------------------------------------------------------
# Test case generation
# ---------------------------------------------------------------------------

TestUrlShortener = UrlShortenerStateMachine.TestCase
TestUrlShortener.settings = settings(
    max_examples=200,
    stateful_step_count=15,
    suppress_health_check=[HealthCheck.too_slow],
)

# file: tests/test_stateful_todo.py
#
# AUTO-GENERATED from spec: TodoList

from hypothesis import settings, HealthCheck
from hypothesis import strategies as st
from hypothesis.stateful import (
    RuleBasedStateMachine,
    Bundle,
    rule,
    invariant,
    precondition,
    initialize,
    multiple,
)
import httpx

BASE_URL = "http://localhost:8000"
client = httpx.Client(base_url=BASE_URL, timeout=10.0)

valid_titles = st.text(
    alphabet=st.characters(whitelist_categories=("L", "N", "P", "Z")),
    min_size=1,
    max_size=200,
).filter(lambda t: len(t.strip()) >= 1)


class TodoItem:
    """Model representation of a todo item, mirrors the spec entity."""
    def __init__(self, id: str, title: str, status: str = "todo"):
        self.id = id
        self.title = title
        self.status = status


class TodoListStateMachine(RuleBasedStateMachine):
    """
    State machine that mirrors the spec's abstract state and enforces
    the state transition constraints:
        todo -> in_progress -> done
    """

    def __init__(self):
        super().__init__()
        self.items: dict[str, TodoItem] = {}

    # Bundles for tracking resources by state
    todo_ids = Bundle("todo_ids")
    in_progress_ids = Bundle("in_progress_ids")
    done_ids = Bundle("done_ids")
    all_ids = Bundle("all_ids")

    @initialize()
    def reset_service(self):
        resp = client.post("/test/reset")
        assert resp.status_code == 200
        self.items = {}

    # -------------------------------------------------------------------
    # OPERATION: CreateTodo
    # ensures: item.id not in pre(items)
    # ensures: item.status = TodoStatus.todo
    # ensures: item.title = title
    # ensures: #items' = #items + 1
    # -------------------------------------------------------------------
    @rule(target=todo_ids, title=valid_titles)
    def create_todo(self, title):
        count_before = len(self.items)

        response = client.post("/todos", json={"title": title})
        assert response.status_code == 201

        data = response.json()
        todo_id = data["id"]

        # ensures: item.id not in pre(items)
        assert todo_id not in self.items

        # ensures: item.status = TodoStatus.todo
        assert data["status"] == "todo"

        # ensures: item.title = title
        assert data["title"] == title

        # Update model
        self.items[todo_id] = TodoItem(id=todo_id, title=title, status="todo")

        # ensures: #items' = #items + 1
        assert len(self.items) == count_before + 1

        return todo_id

    # -------------------------------------------------------------------
    # OPERATION: StartTodo
    # requires: id in items
    # requires: items[id].status = TodoStatus.todo
    # ensures: items'[id].status = TodoStatus.in_progress
    # ensures: #items' = #items
    # -------------------------------------------------------------------
    @rule(target=in_progress_ids, todo_id=todo_ids)
    def start_todo(self, todo_id):
        if todo_id not in self.items:
            return multiple()  # Item was deleted, skip
        if self.items[todo_id].status != "todo":
            return multiple()  # Wrong state, skip

        count_before = len(self.items)

        response = client.patch(f"/todos/{todo_id}/start")
        assert response.status_code == 200

        data = response.json()

        # ensures: items'[id].status = TodoStatus.in_progress
        assert data["status"] == "in_progress"

        # Update model
        self.items[todo_id].status = "in_progress"

        # ensures: #items' = #items
        assert len(self.items) == count_before

        return todo_id

    # -------------------------------------------------------------------
    # OPERATION: CompleteTodo
    # requires: id in items
    # requires: items[id].status = TodoStatus.in_progress
    # ensures: items'[id].status = TodoStatus.done
    # ensures: #items' = #items
    # -------------------------------------------------------------------
    @rule(target=done_ids, ip_id=in_progress_ids)
    def complete_todo(self, ip_id):
        if ip_id not in self.items:
            return multiple()
        if self.items[ip_id].status != "in_progress":
            return multiple()

        count_before = len(self.items)

        response = client.patch(f"/todos/{ip_id}/complete")
        assert response.status_code == 200

        data = response.json()

        # ensures: items'[id].status = TodoStatus.done
        assert data["status"] == "done"

        # Update model
        self.items[ip_id].status = "done"

        # ensures: #items' = #items
        assert len(self.items) == count_before

        return ip_id

    # -------------------------------------------------------------------
    # OPERATION: DeleteTodo
    # requires: id in items
    # ensures: id not in items'
    # ensures: #items' = #items - 1
    # -------------------------------------------------------------------
    @rule(todo_id=todo_ids)
    def delete_todo(self, todo_id):
        if todo_id not in self.items:
            return  # Already deleted

        count_before = len(self.items)

        response = client.delete(f"/todos/{todo_id}")
        assert response.status_code == 204

        # ensures: id not in items'
        del self.items[todo_id]

        # ensures: #items' = #items - 1
        assert len(self.items) == count_before - 1

    # -------------------------------------------------------------------
    # NEGATIVE: Invalid state transitions
    # -------------------------------------------------------------------
    @rule(ip_id=in_progress_ids)
    def start_in_progress_fails(self, ip_id):
        """Cannot start an already in-progress item."""
        if ip_id not in self.items or self.items[ip_id].status != "in_progress":
            return
        response = client.patch(f"/todos/{ip_id}/start")
        assert response.status_code in (409, 422)

    @rule(todo_id=todo_ids)
    def complete_todo_status_fails(self, todo_id):
        """Cannot complete an item that is still in 'todo' status."""
        if todo_id not in self.items or self.items[todo_id].status != "todo":
            return
        response = client.patch(f"/todos/{todo_id}/complete")
        assert response.status_code in (409, 422)

    @rule(done_id=done_ids)
    def start_done_fails(self, done_id):
        """Cannot start a done item (no backward transitions)."""
        if done_id not in self.items or self.items[done_id].status != "done":
            return
        response = client.patch(f"/todos/{done_id}/start")
        assert response.status_code in (409, 422)

    # -------------------------------------------------------------------
    # INVARIANTS
    # -------------------------------------------------------------------
    @invariant()
    def all_titles_non_empty(self):
        """invariant: all t in items | len(items[t].title.value) >= 1"""
        for item in self.items.values():
            assert len(item.title) >= 1

    @invariant()
    def model_matches_service(self):
        """The model and the service must agree on the set of items and their states."""
        resp = client.get("/todos")
        if resp.status_code != 200:
            return
        service_items = {item["id"]: item["status"] for item in resp.json()}
        model_items = {k: v.status for k, v in self.items.items()}
        assert service_items == model_items, (
            f"Model/service mismatch.\n"
            f"  Model:   {model_items}\n"
            f"  Service: {service_items}"
        )

    @invariant()
    def valid_status_values(self):
        """All items have a status that is one of the enum values."""
        for item in self.items.values():
            assert item.status in ("todo", "in_progress", "done"), (
                f"Item {item.id} has invalid status '{item.status}'"
            )


TestTodoList = TodoListStateMachine.TestCase
TestTodoList.settings = settings(
    max_examples=200,
    stateful_step_count=20,
    suppress_health_check=[HealthCheck.too_slow],
)

# file: tests/test_stateful_orders.py
#
# AUTO-GENERATED from spec: OrderSystem

from hypothesis import settings, HealthCheck
from hypothesis import strategies as st
from hypothesis.stateful import (
    RuleBasedStateMachine,
    Bundle,
    rule,
    invariant,
    precondition,
    initialize,
    multiple,
)
import httpx

BASE_URL = "http://localhost:8000"
client = httpx.Client(base_url=BASE_URL, timeout=10.0)

SEED_PRODUCTS = [
    {"id": "prod-001", "name": "Widget", "price_cents": 999},
    {"id": "prod-002", "name": "Gadget", "price_cents": 2499},
    {"id": "prod-003", "name": "Doohickey", "price_cents": 50},
]
CUSTOMER_ID = "cust-001"

order_lines = st.lists(
    st.fixed_dictionaries({
        "product": st.sampled_from([p["id"] for p in SEED_PRODUCTS]),
        "quantity": st.integers(min_value=1, max_value=3),
        "unit_price_cents": st.sampled_from([p["price_cents"] for p in SEED_PRODUCTS]),
    }),
    min_size=1,
    max_size=3,
)


class OrderModel:
    """Model of an order, mirrors the spec entity."""
    def __init__(self, id: str, customer: str, lines: list, status: str, total_cents: int):
        self.id = id
        self.customer = customer
        self.lines = lines
        self.status = status
        self.total_cents = total_cents


class OrderSystemStateMachine(RuleBasedStateMachine):
    """
    State machine for e-commerce order lifecycle.
    Tracks orders and inventory, verifies all spec ensures and invariants.

    State transitions:
        pending -> confirmed -> shipped -> delivered
        pending -> cancelled
        confirmed -> cancelled
    """

    def __init__(self):
        super().__init__()
        self.orders: dict[str, OrderModel] = {}
        self.inventory: dict[str, int] = {}

    # Bundles for tracking orders by status
    pending_ids = Bundle("pending_ids")
    confirmed_ids = Bundle("confirmed_ids")
    shipped_ids = Bundle("shipped_ids")

    @initialize()
    def reset_and_seed(self):
        resp = client.post("/test/reset")
        assert resp.status_code == 200
        self.orders = {}
        self.inventory = {}

        # Seed product catalog
        for p in SEED_PRODUCTS:
            client.put(f"/products/{p['id']}", json={
                "name": p["name"],
                "price_cents": p["price_cents"],
            })
            client.put(f"/products/{p['id']}/inventory", json={"quantity": 100})
            self.inventory[p["id"]] = 100

    # -------------------------------------------------------------------
    # OPERATION: PlaceOrder
    # -------------------------------------------------------------------
    @rule(target=pending_ids, lines=order_lines)
    def place_order(self, lines):
        # Check precondition: sufficient inventory
        needed = {}
        for line in lines:
            needed[line["product"]] = needed.get(line["product"], 0) + line["quantity"]
        for pid, qty in needed.items():
            if self.inventory.get(pid, 0) < qty:
                # Insufficient inventory; expect rejection
                resp = client.post("/orders", json={
                    "customer": CUSTOMER_ID, "lines": lines,
                })
                assert resp.status_code in (409, 422)
                return multiple()

        count_before = len(self.orders)

        resp = client.post("/orders", json={
            "customer": CUSTOMER_ID, "lines": lines,
        })
        assert resp.status_code == 201

        data = resp.json()
        order_id = data["id"]

        # ensures: order.id not in pre(orders)
        assert order_id not in self.orders

        # ensures: order.status = OrderStatus.pending
        assert data["status"] == "pending"

        # ensures: order.total.cents = sum(...)
        expected_total = sum(l["quantity"] * l["unit_price_cents"] for l in lines)
        assert data["total_cents"] == expected_total

        # Update model: store order
        self.orders[order_id] = OrderModel(
            id=order_id,
            customer=CUSTOMER_ID,
            lines=lines,
            status="pending",
            total_cents=expected_total,
        )

        # Update model: decrement inventory
        for line in lines:
            self.inventory[line["product"]] -= line["quantity"]

        # ensures: #orders' = #orders + 1
        assert len(self.orders) == count_before + 1

        return order_id

    # -------------------------------------------------------------------
    # OPERATION: ConfirmOrder
    # requires: orders[id].status = OrderStatus.pending
    # ensures: orders'[id].status = OrderStatus.confirmed
    # -------------------------------------------------------------------
    @rule(target=confirmed_ids, order_id=pending_ids)
    def confirm_order(self, order_id):
        if order_id not in self.orders:
            return multiple()
        if self.orders[order_id].status != "pending":
            return multiple()

        resp = client.patch(f"/orders/{order_id}/confirm")
        assert resp.status_code == 200
        assert resp.json()["status"] == "confirmed"

        self.orders[order_id].status = "confirmed"
        return order_id

    # -------------------------------------------------------------------
    # OPERATION: ShipOrder (implied by spec state machine)
    # requires: orders[id].status = OrderStatus.confirmed
    # ensures: orders'[id].status = OrderStatus.shipped
    # -------------------------------------------------------------------
    @rule(target=shipped_ids, order_id=confirmed_ids)
    def ship_order(self, order_id):
        if order_id not in self.orders:
            return multiple()
        if self.orders[order_id].status != "confirmed":
            return multiple()

        resp = client.patch(f"/orders/{order_id}/ship")
        assert resp.status_code == 200
        assert resp.json()["status"] == "shipped"

        self.orders[order_id].status = "shipped"
        return order_id

    # -------------------------------------------------------------------
    # OPERATION: CancelOrder
    # requires: orders[id].status in {pending, confirmed}
    # ensures: orders'[id].status = OrderStatus.cancelled
    # ensures: inventory restored
    # -------------------------------------------------------------------
    @rule(order_id=pending_ids)
    def cancel_pending_order(self, order_id):
        if order_id not in self.orders:
            return
        if self.orders[order_id].status != "pending":
            return

        resp = client.patch(f"/orders/{order_id}/cancel")
        assert resp.status_code == 200
        assert resp.json()["status"] == "cancelled"

        # Restore inventory in model
        for line in self.orders[order_id].lines:
            self.inventory[line["product"]] += line["quantity"]

        self.orders[order_id].status = "cancelled"

    @rule(order_id=confirmed_ids)
    def cancel_confirmed_order(self, order_id):
        if order_id not in self.orders:
            return
        if self.orders[order_id].status != "confirmed":
            return

        resp = client.patch(f"/orders/{order_id}/cancel")
        assert resp.status_code == 200
        assert resp.json()["status"] == "cancelled"

        # Restore inventory in model
        for line in self.orders[order_id].lines:
            self.inventory[line["product"]] += line["quantity"]

        self.orders[order_id].status = "cancelled"

    # -------------------------------------------------------------------
    # NEGATIVE: Cannot cancel shipped orders
    # -------------------------------------------------------------------
    @rule(order_id=shipped_ids)
    def cancel_shipped_fails(self, order_id):
        if order_id not in self.orders:
            return
        if self.orders[order_id].status != "shipped":
            return
        resp = client.patch(f"/orders/{order_id}/cancel")
        assert resp.status_code in (409, 422)
        # Verify status unchanged
        assert self.orders[order_id].status == "shipped"

    # -------------------------------------------------------------------
    # INVARIANTS
    # -------------------------------------------------------------------
    @invariant()
    def all_totals_non_negative(self):
        """invariant: all o in orders | orders[o].total.cents >= 0"""
        for order in self.orders.values():
            assert order.total_cents >= 0, (
                f"Order {order.id} has negative total: {order.total_cents}"
            )

    @invariant()
    def all_orders_have_lines(self):
        """invariant: all o in orders | #orders[o].lines >= 1"""
        for order in self.orders.values():
            assert len(order.lines) >= 1, (
                f"Order {order.id} has no lines"
            )

    @invariant()
    def inventory_non_negative(self):
        """invariant: all p in inventory | inventory[p] >= 0"""
        for pid, qty in self.inventory.items():
            assert qty >= 0, f"Product {pid} has negative inventory: {qty}"

    @invariant()
    def model_matches_service(self):
        """The service's order statuses must match the model."""
        resp = client.get("/orders")
        if resp.status_code != 200:
            return
        service_orders = {o["id"]: o["status"] for o in resp.json()}
        model_orders = {k: v.status for k, v in self.orders.items()}
        assert service_orders == model_orders, (
            f"Model/service order status mismatch.\n"
            f"  Model:   {model_orders}\n"
            f"  Service: {service_orders}"
        )


TestOrderSystem = OrderSystemStateMachine.TestCase
TestOrderSystem.settings = settings(
    max_examples=150,
    stateful_step_count=25,
    suppress_health_check=[HealthCheck.too_slow],
)