......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
import os
import secrets
import threading
from typing import Optional
from fastapi import FastAPI, Header, Query, Request
from fastapi.responses import JSONResponse
# ---------------------------------------------------------------------------
# In-memory data store
# ---------------------------------------------------------------------------
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = [
{"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
{"id": 2, "name": "Mechanical Keyboard", "price": 89.50, "stock": 7},
{"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 15},
]
ORDERS: list[dict] = []
NEXT_ORDER_ID = 1
# Token store: token_string -> username
TOKENS: dict[str, str] = {}
# Lock for thread safety
_lock = threading.Lock()
# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _json_error(status: int, message: str) -> JSONResponse:
return JSONResponse(status_code=status, content={"error": message})
def _authenticate(authorization: Optional[str]) -> tuple[bool, str]:
"""Return (ok, username)."""
if not authorization or not authorization.startswith("Bearer "):
return False, ""
token = authorization[len("Bearer "):]
with _lock:
username = TOKENS.get(token)
if username is None:
return False, ""
return True, username
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return _json_error(400, "Malformed JSON body")
if not isinstance(body, dict):
return _json_error(400, "Request body must be a JSON object")
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return _json_error(400, "Missing 'username' or 'password'")
if not isinstance(username, str) or not isinstance(password, str):
return _json_error(400, "'username' and 'password' must be strings")
if USERS.get(username) != password:
return _json_error(401, "Invalid credentials")
token = secrets.token_urlsafe(32)
with _lock:
TOKENS[token] = username
return JSONResponse(status_code=200, content={"token": token})
@app.get("/products")
async def list_products(max_price: Optional[str] = Query(None)):
products = PRODUCTS[:]
if max_price is not None:
try:
limit = float(max_price)
except (ValueError, TypeError):
pass # invalid max_price -> return all
else:
products = [p for p in products if p["price"] <= limit]
return JSONResponse(status_code=200, content=products)
@app.get("/products/{product_id}")
async def get_product(product_id: int):
for p in PRODUCTS:
if p["id"] == product_id:
return JSONResponse(status_code=200, content=p)
return _json_error(404, f"Product {product_id} not found")
@app.post("/orders")
async def create_order(request: Request, authorization: Optional[str] = Header(None)):
ok, username = _authenticate(authorization)
if not ok:
return _json_error(401, "Invalid or missing token")
try:
body = await request.json()
except Exception:
return _json_error(400, "Malformed JSON body")
if not isinstance(body, dict):
return _json_error(400, "Request body must be a JSON object")
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
return _json_error(400, "Missing 'product_id' or 'quantity'")
if not isinstance(product_id, int) or isinstance(product_id, bool):
return _json_error(400, "'product_id' must be an integer")
if not isinstance(quantity, int) or isinstance(quantity, bool):
return _json_error(400, "'quantity' must be an integer")
if quantity < 1:
return _json_error(400, "'quantity' must be at least 1")
global NEXT_ORDER_ID
with _lock:
# Find product
product = None
for p in PRODUCTS:
if p["id"] == product_id:
product = p
break
if product is None:
return _json_error(404, f"Product {product_id} not found")
if quantity > product["stock"]:
return _json_error(409, "Insufficient stock")
product["stock"] -= quantity
total = product["price"] * quantity
order_id = NEXT_ORDER_ID
NEXT_ORDER_ID += 1
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
}
ORDERS.append(order)
return JSONResponse(status_code=201, content={
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
})
@app.get("/orders")
async def list_orders(authorization: Optional[str] = Header(None)):
ok, username = _authenticate(authorization)
if not ok:
return _json_error(401, "Invalid or missing token")
with _lock:
user_orders = [
{k: v for k, v in o.items() if k != "username"}
for o in ORDERS
if o["username"] == username
]
return JSONResponse(status_code=200, content=user_orders)
@app.get("/orders/{order_id}")
async def get_order(order_id: int, authorization: Optional[str] = Header(None)):
ok, username = _authenticate(authorization)
if not ok:
return _json_error(401, "Invalid or missing token")
with _lock:
for o in ORDERS:
if o["id"] == order_id:
if o["username"] != username:
return _json_error(404, "Order not found")
return JSONResponse(status_code=200, content={
"id": o["id"],
"product_id": o["product_id"],
"quantity": o["quantity"],
"total": o["total"],
})
return _json_error(404, "Order not found")
# ---------------------------------------------------------------------------
# Main (uvicorn)
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
"""Online Store REST API — in-memory, FastAPI."""
import os
import secrets
from typing import Optional
from fastapi import FastAPI, Header, Query, Request
from fastapi.responses import JSONResponse
# ---------------------------------------------------------------------------
# In-memory stores
# ---------------------------------------------------------------------------
USERS: dict[str, str] = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS: dict[int, dict] = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 89.50, "stock": 7},
3: {"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 15},
}
ORDERS: dict[int, dict] = {}
_next_order_id: int = 1
TOKENS: dict[str, str] = {} # token -> username
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _json_error(status: int, message: str) -> JSONResponse:
return JSONResponse(status_code=status, content={"error": message})
def _parse_token(authorization: Optional[str]) -> Optional[str]:
"""Return the username if the bearer token is valid, else None."""
if not authorization or not authorization.startswith("Bearer "):
return None
token = authorization[len("Bearer "):]
return TOKENS.get(token)
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastAPI()
@app.exception_handler(Exception)
async def _generic_handler(request: Request, exc: Exception):
return _json_error(500, "Internal server error")
# ---- Public endpoints -----------------------------------------------------
@app.get("/products")
async def list_products(max_price: Optional[str] = Query(None)):
results = list(PRODUCTS.values())
if max_price is not None:
try:
limit = float(max_price)
except (ValueError, TypeError):
pass # invalid → return all
else:
results = [p for p in results if p["price"] <= limit]
return results
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
return _json_error(404, f"Product {product_id} not found")
return product
# ---- Auth -----------------------------------------------------------------
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return _json_error(400, "Body must be valid JSON")
if not isinstance(body, dict):
return _json_error(400, "Body must be a JSON object")
username = body.get("username")
password = body.get("password")
if not username or not password:
return _json_error(400, "Missing 'username' or 'password'")
if username not in USERS or USERS[username] != password:
return _json_error(401, "Invalid credentials")
token = secrets.token_urlsafe(32)
TOKENS[token] = username
return {"token": token}
# ---- Protected endpoints --------------------------------------------------
def _check_auth(authorization: Optional[str]) -> tuple[str, Optional[JSONResponse]]:
"""Return (username, None) on success or ('', error_response) on failure."""
user = _parse_token(authorization)
if user is None:
return "", _json_error(401, "Missing or invalid token")
return user, None
@app.post("/orders")
async def create_order(
request: Request,
authorization: Optional[str] = Header(None),
):
user, err = _check_auth(authorization)
if err is not None:
return err
# Parse body
try:
body = await request.json()
except Exception:
return _json_error(400, "Body must be valid JSON")
if not isinstance(body, dict):
return _json_error(400, "Body must be a JSON object")
product_id = body.get("product_id")
quantity = body.get("quantity")
# Validate presence and types
if product_id is None or quantity is None:
return _json_error(400, "Missing 'product_id' or 'quantity'")
if not isinstance(product_id, int) or isinstance(product_id, bool):
return _json_error(400, "'product_id' must be an integer")
if not isinstance(quantity, int) or isinstance(quantity, bool):
return _json_error(400, "'quantity' must be an integer")
if quantity < 1:
return _json_error(400, "'quantity' must be at least 1")
# Look up product
product = PRODUCTS.get(product_id)
if product is None:
return _json_error(404, f"Product {product_id} not found")
# Check stock
if quantity > product["stock"]:
return _json_error(409, "Not enough stock")
# Create order
global _next_order_id
order_id = _next_order_id
_next_order_id += 1
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"user": user,
}
ORDERS[order_id] = order
# Reduce stock
product["stock"] -= quantity
return JSONResponse(status_code=201, content=order)
@app.get("/orders")
async def list_orders(
authorization: Optional[str] = Header(None),
):
user, err = _check_auth(authorization)
if err is not None:
return err
my_orders = [
{k: v for k, v in o.items() if k != "user"}
for o in ORDERS.values()
if o["user"] == user
]
return my_orders
@app.get("/orders/{order_id}")
async def get_order(
order_id: int,
authorization: Optional[str] = Header(None),
):
user, err = _check_auth(authorization)
if err is not None:
return err
order = ORDERS.get(order_id)
if order is None or order["user"] != user:
return _json_error(404, "Order not found")
return {k: v for k, v in order.items() if k != "user"}
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run("app:app", host=host, port=port, reload=False)
fastapi uvicorn
............FFFFFF..F.FF. [100%]
=================================== FAILURES ===================================
__________________________ test_login_malformed_json ___________________________
base_url = 'http://127.0.0.1:42577'
def test_login_malformed_json(base_url):
status, parsed, ct = _request_h("POST", f"{base_url}/login", raw_body="{not valid json")
> _assert_json_error(status, parsed, ct, 400)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:261:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 422
parsed = {'detail': [{'ctx': {'error': 'Expecting property name enclosed in double quotes'}, 'input': {}, 'loc': ['body', 1], 'msg': 'JSON decode error', ...}]}
ct = 'application/json', expect_status = 400
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
> assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
E AssertionError: expected 400, got 422 (body={'detail': [{'type': 'json_invalid', 'loc': ['body', 1], 'msg': 'JSON decode error', 'input': {}, 'ctx': {'error': 'Expecting property name enclosed in double quotes'}}]})
E assert 422 == 400
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:89: AssertionError
__________________________ test_login_missing_fields ___________________________
base_url = 'http://127.0.0.1:42577'
def test_login_missing_fields(base_url):
status, parsed, ct = _request_h("POST", f"{base_url}/login", body={"username": "alice"})
> _assert_json_error(status, parsed, ct, 400)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:266:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 422
parsed = {'detail': [{'input': {'username': 'alice'}, 'loc': ['body', 'password'], 'msg': 'Field required', 'type': 'missing'}]}
ct = 'application/json', expect_status = 400
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
> assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
E AssertionError: expected 400, got 422 (body={'detail': [{'type': 'missing', 'loc': ['body', 'password'], 'msg': 'Field required', 'input': {'username': 'alice'}}]})
E assert 422 == 400
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:89: AssertionError
__________________________ test_order_malformed_json ___________________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_order_malformed_json(base_url, alice_token):
status, parsed, ct = _request_h("POST", f"{base_url}/orders", token=alice_token,
raw_body="definitely not json")
> _assert_json_error(status, parsed, ct, 400)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:272:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 400, parsed = {'detail': {'error': 'Body must be valid JSON.'}}
ct = 'application/json', expect_status = 400
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'Body must be valid JSON.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Body must be valid JSON.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f20c00>('error')
E + where <built-in method get of dict object at 0x7fc459f20c00> = {'detail': {'error': 'Body must be valid JSON.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f20c00>('message')
E + where <built-in method get of dict object at 0x7fc459f20c00> = {'detail': {'error': 'Body must be valid JSON.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
__________________________ test_order_missing_fields ___________________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_order_missing_fields(base_url, alice_token):
status, parsed, ct = _request_h("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1})
> _assert_json_error(status, parsed, ct, 400)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:278:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 400
parsed = {'detail': {'error': 'Missing required fields: product_id and quantity.'}}
ct = 'application/json', expect_status = 400
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'Missing required fields: product_id and quantity.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Missing required fields: product_id and quantity.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f83cc0>('error')
E + where <built-in method get of dict object at 0x7fc459f83cc0> = {'detail': {'error': 'Missing required fields: product_id and quantity.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f83cc0>('message')
E + where <built-in method get of dict object at 0x7fc459f83cc0> = {'detail': {'error': 'Missing required fields: product_id and quantity.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
________________________ test_order_non_integer_fields _________________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_order_non_integer_fields(base_url, alice_token):
status, parsed, ct = _request_h("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": "abc", "quantity": "two"})
> _assert_json_error(status, parsed, ct, 400)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:284:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 400, parsed = {'detail': {'error': 'product_id must be an integer.'}}
ct = 'application/json', expect_status = 400
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'product_id must be an integer.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'product_id must be an integer.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f7b480>('error')
E + where <built-in method get of dict object at 0x7fc459f7b480> = {'detail': {'error': 'product_id must be an integer.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f7b480>('message')
E + where <built-in method get of dict object at 0x7fc459f7b480> = {'detail': {'error': 'product_id must be an integer.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
_____________________ test_order_zero_or_negative_quantity _____________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_order_zero_or_negative_quantity(base_url, alice_token):
status, parsed, ct = _request_h("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1, "quantity": 0})
> _assert_json_error(status, parsed, ct, 400)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:290:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 400, parsed = {'detail': {'error': 'quantity must be at least 1.'}}
ct = 'application/json', expect_status = 400
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'quantity must be at least 1.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'quantity must be at least 1.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f8fd80>('error')
E + where <built-in method get of dict object at 0x7fc459f8fd80> = {'detail': {'error': 'quantity must be at least 1.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f8fd80>('message')
E + where <built-in method get of dict object at 0x7fc459f8fd80> = {'detail': {'error': 'quantity must be at least 1.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
_____________________ test_order_exceeding_stock_conflict ______________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_order_exceeding_stock_conflict(base_url, alice_token):
_, before, _ = _request("GET", f"{base_url}/products/1")
start = before["stock"]
status, parsed, ct = _request_h("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1, "quantity": start + 1000})
> _assert_json_error(status, parsed, ct, 409)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:318:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 409
parsed = {'detail': {'error': 'Insufficient stock. Available: 7.'}}
ct = 'application/json', expect_status = 409
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'Insufficient stock. Available: 7.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Insufficient stock. Available: 7.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f6b500>('error')
E + where <built-in method get of dict object at 0x7fc459f6b500> = {'detail': {'error': 'Insufficient stock. Available: 7.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f6b500>('message')
E + where <built-in method get of dict object at 0x7fc459f6b500> = {'detail': {'error': 'Insufficient stock. Available: 7.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
____________________________ test_get_missing_order ____________________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_get_missing_order(base_url, alice_token):
status, parsed, ct = _request_h("GET", f"{base_url}/orders/999999", token=alice_token)
> _assert_json_error(status, parsed, ct, 404)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:337:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 404, parsed = {'detail': {'error': 'Order not found.'}}
ct = 'application/json', expect_status = 404
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'Order not found.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Order not found.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f87240>('error')
E + where <built-in method get of dict object at 0x7fc459f87240> = {'detail': {'error': 'Order not found.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f87240>('message')
E + where <built-in method get of dict object at 0x7fc459f87240> = {'detail': {'error': 'Order not found.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
______________________ test_get_other_users_order_is_404 _______________________
base_url = 'http://127.0.0.1:42577'
alice_token = '232a59c2c3d54cf14bbe0d58575023fead136ddfcceaebdb72a64397517cf94e'
def test_get_other_users_order_is_404(base_url, alice_token):
# alice creates an order
_, order, _ = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 2, "quantity": 1})
oid = order["id"]
# bob must not be able to read it (404, not 403 — don't reveal it exists)
_, blogin, _ = _request("POST", f"{base_url}/login",
body={"username": "bob", "password": "hunter2"})
bob_token = blogin["token"]
status, parsed, ct = _request_h("GET", f"{base_url}/orders/{oid}", token=bob_token)
> _assert_json_error(status, parsed, ct, 404)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:350:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 404, parsed = {'detail': {'error': 'Order not found.'}}
ct = 'application/json', expect_status = 404
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': {'error': 'Order not found.'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Order not found.'}}, dict)
E + and None = <built-in method get of dict object at 0x7fc459f65940>('error')
E + where <built-in method get of dict object at 0x7fc459f65940> = {'detail': {'error': 'Order not found.'}}.get
E + and None = <built-in method get of dict object at 0x7fc459f65940>('message')
E + where <built-in method get of dict object at 0x7fc459f65940> = {'detail': {'error': 'Order not found.'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
=========================== short test summary info ============================
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_malformed_json
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_missing_fields
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_malformed_json
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_missing_fields
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_non_integer_fields
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_zero_or_negative_quantity
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_exceeding_stock_conflict
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_missing_order
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_other_users_order_is_404
9 failed, 16 passed in 0.71s
score: 16/25 (64%)
import os
import secrets
from typing import Optional
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
app = FastAPI()
# ── In-memory data ──────────────────────────────────────────────────────────
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = [
{"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
{"id": 2, "name": "Mechanical Keyboard", "price": 89.50, "stock": 7},
{"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 15},
]
ORDERS = [] # list of dicts
NEXT_ORDER_ID = 1
# token -> username
TOKENS: dict[str, str] = {}
# ── Helpers ─────────────────────────────────────────────────────────────────
def _get_user_from_token(auth_header: Optional[str]) -> str:
"""Return username if token is valid, otherwise raise 401."""
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail={"error": "Missing or invalid Authorization header."})
token = auth_header[len("Bearer "):]
username = TOKENS.get(token)
if username is None:
raise HTTPException(status_code=401, detail={"error": "Invalid token."})
return username
def _find_product(product_id: int):
for p in PRODUCTS:
if p["id"] == product_id:
return p
return None
# ── Public endpoints ────────────────────────────────────────────────────────
@app.get("/products")
async def list_products(max_price: Optional[str] = None):
results = PRODUCTS[:]
if max_price is not None:
try:
threshold = float(max_price)
except (ValueError, TypeError):
pass # invalid max_price -> return all
else:
results = [p for p in results if p["price"] <= threshold]
return results
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = _find_product(product_id)
if product is None:
raise HTTPException(status_code=404, detail={"error": f"Product {product_id} not found."})
return product
# ── Auth endpoint ───────────────────────────────────────────────────────────
class LoginBody(BaseModel):
username: str
password: str
@app.post("/login")
async def login(body: LoginBody):
if body.password == USERS.get(body.username):
token = secrets.token_hex(32)
TOKENS[token] = body.username
return {"token": token}
raise HTTPException(status_code=401, detail={"error": "Invalid credentials."})
# ── Protected endpoints ─────────────────────────────────────────────────────
@app.post("/orders")
async def create_order(request: Request, authorization: Optional[str] = Header(None)):
username = _get_user_from_token(authorization)
# Parse body
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Body must be valid JSON."})
if "product_id" not in data or "quantity" not in data:
raise HTTPException(status_code=400, detail={"error": "Missing required fields: product_id and quantity."})
product_id = data["product_id"]
quantity = data["quantity"]
# Validate types: must be integers
if not isinstance(product_id, int) or isinstance(product_id, bool):
raise HTTPException(status_code=400, detail={"error": "product_id must be an integer."})
if not isinstance(quantity, int) or isinstance(quantity, bool):
raise HTTPException(status_code=400, detail={"error": "quantity must be an integer."})
if quantity < 1:
raise HTTPException(status_code=400, detail={"error": "quantity must be at least 1."})
# Find product
product = _find_product(product_id)
if product is None:
raise HTTPException(status_code=404, detail={"error": f"Product {product_id} not found."})
# Check stock
if quantity > product["stock"]:
raise HTTPException(
status_code=409,
detail={"error": f"Insufficient stock. Available: {product['stock']}."},
)
# Create order
global NEXT_ORDER_ID
total = product["price"] * quantity
order = {
"id": NEXT_ORDER_ID,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
}
NEXT_ORDER_ID += 1
ORDERS.append(order)
# Reduce stock
product["stock"] -= quantity
return JSONResponse(status_code=201, content={
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
})
@app.get("/orders")
async def list_my_orders(authorization: Optional[str] = Header(None)):
username = _get_user_from_token(authorization)
my_orders = [
{k: v for k, v in o.items() if k != "username"}
for o in ORDERS
if o["username"] == username
]
return my_orders
@app.get("/orders/{order_id}")
async def get_my_order(order_id: int, authorization: Optional[str] = Header(None)):
username = _get_user_from_token(authorization)
for o in ORDERS:
if o["id"] == order_id:
if o["username"] != username:
raise HTTPException(status_code=404, detail={"error": "Order not found."})
return {k: v for k, v in o.items() if k != "username"}
raise HTTPException(status_code=404, detail={"error": "Order not found."})
# ── Entry point ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
"""Online store REST API."""
import os
import secrets
from typing import Optional
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request
from fastapi.responses import JSONResponse
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS: dict[int, dict] = {
1: {"id": 1, "name": "Widget", "price": 9.99, "stock": 10},
2: {"id": 2, "name": "Gadget", "price": 24.50, "stock": 8},
3: {"id": 3, "name": "Doohickey", "price": 4.75, "stock": 20},
}
# token -> username
TOKENS: dict[str, str] = {}
# order_id -> order dict (each order also stores the username that created it)
ORDERS: dict[int, dict] = {}
_next_order_id = 1
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _ok(data, status: int = 200) -> JSONResponse:
return JSONResponse(content=data, status_code=status)
def _err(message: str, status: int = 400) -> JSONResponse:
return JSONResponse(content={"error": message}, status_code=status)
def _parse_product_id(raw) -> int:
"""Validate that raw is an integer (not bool)."""
if isinstance(raw, bool) or not isinstance(raw, int):
raise HTTPException(status_code=400, detail="product_id must be an integer.")
return raw
def _parse_quantity(raw) -> int:
if isinstance(raw, bool) or not isinstance(raw, int):
raise HTTPException(status_code=400, detail="quantity must be an integer.")
if raw < 1:
raise HTTPException(status_code=400, detail="quantity must be at least 1.")
return raw
# ---------------------------------------------------------------------------
# Auth dependency
# ---------------------------------------------------------------------------
async def get_current_username(authorization: Optional[str] = Header(default=None)) -> str:
"""Dependency that extracts and validates the Bearer token, returning the username."""
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid authorization header.")
token = authorization[len("Bearer "):]
username = TOKENS.get(token)
if username is None:
raise HTTPException(status_code=401, detail="Invalid token.")
return username
# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="Online Store API")
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.get("/products")
async def list_products(max_price: Optional[str] = Query(default=None)):
"""Return all products, optionally filtered by max_price."""
result = list(PRODUCTS.values())
if max_price is not None:
try:
threshold = float(max_price)
result = [p for p in result if p["price"] <= threshold]
except (ValueError, TypeError):
pass # invalid max_price -> return all
return _ok(result)
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
return _err("Product not found.", 404)
return _ok(product)
# ---------------------------------------------------------------------------
# Authenticated endpoints
# ---------------------------------------------------------------------------
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return _err("Request body must be valid JSON.", 400)
if not isinstance(body, dict):
return _err("Request body must be a JSON object.", 400)
username = body.get("username")
password = body.get("password")
if not username or not password:
return _err("Missing username or password.", 400)
if not isinstance(username, str) or not isinstance(password, str):
return _err("username and password must be strings.", 400)
if USERS.get(username) != password:
return _err("Invalid credentials.", 401)
token = secrets.token_urlsafe(32)
TOKENS[token] = username
return _ok({"token": token})
@app.post("/orders")
async def create_order(
request: Request,
username: str = Depends(get_current_username),
):
# Parse body
try:
body = await request.json()
except Exception:
return _err("Request body must be valid JSON.", 400)
if not isinstance(body, dict):
return _err("Request body must be a JSON object.", 400)
raw_pid = body.get("product_id")
raw_qty = body.get("quantity")
if raw_pid is None or raw_qty is None:
return _err("Missing product_id or quantity.", 400)
try:
product_id = _parse_product_id(raw_pid)
quantity = _parse_quantity(raw_qty)
except HTTPException as exc:
return JSONResponse(content={"error": exc.detail}, status_code=exc.status_code)
product = PRODUCTS.get(product_id)
if product is None:
return _err("Product not found.", 404)
if quantity > product["stock"]:
return _err("Insufficient stock.", 409)
global _next_order_id
order_id = _next_order_id
_next_order_id += 1
total = product["price"] * quantity
product["stock"] -= quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
}
ORDERS[order_id] = order
return _ok({
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}, 201)
@app.get("/orders")
async def list_orders(username: str = Depends(get_current_username)):
user_orders = [
{k: v for k, v in o.items() if k != "username"}
for o in ORDERS.values()
if o["username"] == username
]
return _ok(user_orders)
@app.get("/orders/{order_id}")
async def get_order(order_id: int, username: str = Depends(get_current_username)):
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
return _err("Order not found.", 404)
return _ok({k: v for k, v in order.items() if k != "username"})
# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
import os
import uuid
from fastapi import FastAPI, Request, Query, Header
from fastapi.responses import JSONResponse
from pydantic import BaseModel
app = FastAPI()
# ── In-memory data stores ──────────────────────────────────────────────
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = {
1: {"id": 1, "name": "Widget", "price": 9.99, "stock": 10},
2: {"id": 2, "name": "Gadget", "price": 24.50, "stock": 8},
3: {"id": 3, "name": "Doohickey", "price": 4.75, "stock": 15},
}
ORDERS = {} # order_id -> order dict
TOKEN_STORE = {} # token -> username
_next_order_id = 1
# ── Helpers ────────────────────────────────────────────────────────────
def json_error(status: int, message: str) -> JSONResponse:
return JSONResponse(status_code=status, content={"error": message})
def authenticate(authorization: str | None) -> str | None:
"""Return username if token is valid, else None."""
if not authorization:
return None
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
return None
token = parts[1]
return TOKEN_STORE.get(token)
# ── Public endpoints ───────────────────────────────────────────────────
@app.get("/products")
async def list_products(max_price: str | None = Query(default=None)):
products = list(PRODUCTS.values())
if max_price is not None:
try:
limit = float(max_price)
except (ValueError, TypeError):
pass # invalid → return all
else:
products = [p for p in products if p["price"] <= limit]
return products
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
return json_error(404, "Product not found")
return product
# ── Auth endpoint ──────────────────────────────────────────────────────
class LoginBody(BaseModel):
username: str
password: str
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return json_error(400, "Invalid JSON body")
if not isinstance(body, dict) or "username" not in body or "password" not in body:
return json_error(400, "Missing username or password")
username = body["username"]
password = body["password"]
if not isinstance(username, str) or not isinstance(password, str):
return json_error(400, "Username and password must be strings")
if USERS.get(username) != password:
return json_error(401, "Invalid credentials")
token = uuid.uuid4().hex
TOKEN_STORE[token] = username
return {"token": token}
# ── Protected endpoints ────────────────────────────────────────────────
@app.post("/orders")
async def create_order(request: Request, authorization: str | None = Header(default=None)):
username = authenticate(authorization)
if username is None:
return json_error(401, "Missing or invalid token")
try:
body = await request.json()
except Exception:
return json_error(400, "Invalid JSON body")
if not isinstance(body, dict):
return json_error(400, "Request body must be a JSON object")
if "product_id" not in body or "quantity" not in body:
return json_error(400, "Missing product_id or quantity")
product_id = body["product_id"]
quantity = body["quantity"]
# Validate integer types (reject floats like 2.0)
if not isinstance(product_id, int) or isinstance(product_id, bool):
return json_error(400, "product_id must be an integer")
if not isinstance(quantity, int) or isinstance(quantity, bool):
return json_error(400, "quantity must be an integer")
if quantity < 1:
return json_error(400, "quantity must be at least 1")
product = PRODUCTS.get(product_id)
if product is None:
return json_error(404, "Product not found")
if quantity > product["stock"]:
return json_error(409, "Insufficient stock")
global _next_order_id
order_id = _next_order_id
_next_order_id += 1
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
}
ORDERS[order_id] = order
product["stock"] -= quantity
# Return without the internal 'username' field
return JSONResponse(
status_code=201,
content={
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
},
)
@app.get("/orders")
async def list_orders(authorization: str | None = Header(default=None)):
username = authenticate(authorization)
if username is None:
return json_error(401, "Missing or invalid token")
result = []
for order in ORDERS.values():
if order["username"] == username:
result.append({
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
})
return result
@app.get("/orders/{order_id}")
async def get_order(order_id: int, authorization: str | None = Header(default=None)):
username = authenticate(authorization)
if username is None:
return json_error(401, "Missing or invalid token")
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
return json_error(404, "Order not found")
return {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}
# ── Entry point ────────────────────────────────────────────────────────
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn