............FFFFFF..F.FF. [100%]
=================================== FAILURES ===================================
__________________________ test_login_malformed_json ___________________________
base_url = 'http://127.0.0.1:42851'
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 = 400, parsed = {'detail': 'Invalid JSON body'}, 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': 'Invalid JSON body'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Invalid JSON body'}, dict)
E + and None = <built-in method get of dict object at 0x7f5874220040>('error')
E + where <built-in method get of dict object at 0x7f5874220040> = {'detail': 'Invalid JSON body'}.get
E + and None = <built-in method get of dict object at 0x7f5874220040>('message')
E + where <built-in method get of dict object at 0x7f5874220040> = {'detail': 'Invalid JSON body'}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
__________________________ test_login_missing_fields ___________________________
base_url = 'http://127.0.0.1:42851'
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 = 400, parsed = {'detail': 'Missing username or password'}
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': 'Missing username or password'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Missing username or password'}, dict)
E + and None = <built-in method get of dict object at 0x7f5874022440>('error')
E + where <built-in method get of dict object at 0x7f5874022440> = {'detail': 'Missing username or password'}.get
E + and None = <built-in method get of dict object at 0x7f5874022440>('message')
E + where <built-in method get of dict object at 0x7f5874022440> = {'detail': 'Missing username or password'}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
__________________________ test_order_malformed_json ___________________________
base_url = 'http://127.0.0.1:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': 'Invalid JSON body'}, 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': 'Invalid JSON body'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Invalid JSON body'}, dict)
E + and None = <built-in method get of dict object at 0x7f587405c980>('error')
E + where <built-in method get of dict object at 0x7f587405c980> = {'detail': 'Invalid JSON body'}.get
E + and None = <built-in method get of dict object at 0x7f587405c980>('message')
E + where <built-in method get of dict object at 0x7f587405c980> = {'detail': 'Invalid JSON body'}.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:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': 'Missing product_id or 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': 'Missing product_id or quantity'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Missing product_id or quantity'}, dict)
E + and None = <built-in method get of dict object at 0x7f58740afd40>('error')
E + where <built-in method get of dict object at 0x7f58740afd40> = {'detail': 'Missing product_id or quantity'}.get
E + and None = <built-in method get of dict object at 0x7f58740afd40>('message')
E + where <built-in method get of dict object at 0x7f58740afd40> = {'detail': 'Missing product_id or 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:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': 'product_id and quantity must be integers'}
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': 'product_id and quantity must be integers'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'product_id and quantity must be integers'}, dict)
E + and None = <built-in method get of dict object at 0x7f5874222a00>('error')
E + where <built-in method get of dict object at 0x7f5874222a00> = {'detail': 'product_id and quantity must be integers'}.get
E + and None = <built-in method get of dict object at 0x7f5874222a00>('message')
E + where <built-in method get of dict object at 0x7f5874222a00> = {'detail': 'product_id and quantity must be integers'}.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:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': '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': 'quantity must be at least 1'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'quantity must be at least 1'}, dict)
E + and None = <built-in method get of dict object at 0x7f5874061f40>('error')
E + where <built-in method get of dict object at 0x7f5874061f40> = {'detail': 'quantity must be at least 1'}.get
E + and None = <built-in method get of dict object at 0x7f5874061f40>('message')
E + where <built-in method get of dict object at 0x7f5874061f40> = {'detail': '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:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': 'Insufficient stock'}, 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': 'Insufficient stock'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Insufficient stock'}, dict)
E + and None = <built-in method get of dict object at 0x7f58740d7a00>('error')
E + where <built-in method get of dict object at 0x7f58740d7a00> = {'detail': 'Insufficient stock'}.get
E + and None = <built-in method get of dict object at 0x7f58740d7a00>('message')
E + where <built-in method get of dict object at 0x7f58740d7a00> = {'detail': 'Insufficient stock'}.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:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': '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': 'Order not found'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Order not found'}, dict)
E + and None = <built-in method get of dict object at 0x7f58740c7440>('error')
E + where <built-in method get of dict object at 0x7f58740c7440> = {'detail': 'Order not found'}.get
E + and None = <built-in method get of dict object at 0x7f58740c7440>('message')
E + where <built-in method get of dict object at 0x7f58740c7440> = {'detail': '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:42851'
alice_token = 'd72078b5211299a5602369f9ecd71d77'
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': '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': 'Order not found'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Order not found'}, dict)
E + and None = <built-in method get of dict object at 0x7f58740affc0>('error')
E + where <built-in method get of dict object at 0x7f58740affc0> = {'detail': 'Order not found'}.get
E + and None = <built-in method get of dict object at 0x7f58740affc0>('message')
E + where <built-in method get of dict object at 0x7f58740affc0> = {'detail': '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.67s
score: 16/25 (64%)
from fastapi import FastAPI, HTTPException, Header, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import os
import secrets
app = FastAPI()
# In-memory data storage
users = {
"alice": "password123",
"bob": "hunter2"
}
products = [
{"id": 1, "name": "Product 1", "price": 10.0, "stock": 5},
{"id": 2, "name": "Product 2", "price": 20.0, "stock": 8},
{"id": 3, "name": "Product 3", "price": 30.0, "stock": 10}
]
# Simple in-memory token store (token -> username)
valid_tokens: Dict[str, str] = {}
# Order ID counter
order_id_counter = 1
# Order storage (order_id -> order)
orders: Dict[int, Dict[str, Any]] = {}
# Helper function to get a user from token
def get_user_from_token(authorization: Optional[str] = None) -> str:
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid token")
token = authorization.split("Bearer ")[1]
username = valid_tokens.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except:
raise HTTPException(status_code=400, detail="Invalid JSON body")
username = body.get("username")
password = body.get("password")
if not username or not password:
raise HTTPException(status_code=400, detail="Missing username or password")
if users.get(username) == password:
# Generate a token
token = secrets.token_hex(16)
valid_tokens[token] = username
return JSONResponse(content={"token": token})
else:
raise HTTPException(status_code=401, detail="Invalid credentials")
@app.get("/products")
async def get_products(max_price: Optional[float] = None):
filtered_products = products
if max_price is not None:
try:
max_price = float(max_price)
filtered_products = [p for p in products if p["price"] <= max_price]
except ValueError:
filtered_products = products
return JSONResponse(content=filtered_products)
@app.get("/products/{id}")
async def get_product(id: int):
for product in products:
if product["id"] == id:
return JSONResponse(content=product)
raise HTTPException(status_code=404, detail="Product not found")
@app.put("/products/{id}")
async def update_product(id: int, request: Request):
body = await request.json()
for product in products:
if product["id"] == id:
product["stock"] = body.get("stock", product["stock"])
return JSONResponse(content=product)
raise HTTPException(status_code=404, detail="Product not found")
@app.post("/orders")
async def create_order(
request: Request,
authorization: Optional[str] = Header(None, alias="Authorization")
):
username = get_user_from_token(authorization)
try:
body = await request.json()
except:
raise HTTPException(status_code=400, detail="Invalid JSON body")
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
raise HTTPException(status_code=400, detail="Missing product_id or quantity")
if not isinstance(product_id, int) or not isinstance(quantity, int):
raise HTTPException(status_code=400, detail="product_id and quantity must be integers")
if quantity < 1:
raise HTTPException(status_code=400, detail="quantity must be at least 1")
# Find the product
product = None
for p in products:
if p["id"] == product_id:
product = p
break
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if product["stock"] < quantity:
raise HTTPException(status_code=409, detail="Insufficient stock")
# Create the order
global order_id_counter
order = {
"id": order_id_counter,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
"username": username
}
orders[order_id_counter] = order
# Update product stock
product["stock"] -= quantity
order_id_counter += 1
return JSONResponse(status_code=201, content=order)
@app.get("/orders")
async def get_orders(
authorization: Optional[str] = Header(None, alias="Authorization")
):
username = get_user_from_token(authorization)
user_orders = [order for order in orders.values() if order["username"] == username]
return JSONResponse(content=user_orders)
@app.get("/orders/{id}")
async def get_order(
id: int,
authorization: Optional[str] = Header(None, alias="Authorization")
):
username = get_user_from_token(authorization)
order = orders.get(id)
if not order or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return JSONResponse(content=order)
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)
#!/usr/bin/env python3
import subprocess
import time
import requests
import json
import os
import signal
def test_api():
base_url = "http://127.0.0.1:8000"
print("\n=== COMPREHENSIVE API TEST ===\n")
# Helper function to start server with fresh state
def start_server():
print("Starting the server...")
# Set environment variables to ensure fresh state
env = os.environ.copy()
env['PYTHONPATH'] = '/work'
server_process = subprocess.Popen(
["python3", "app.py"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env
)
time.sleep(3)
return server_process
# Test 1: Test public endpoints
print("1. Testing GET /products (public endpoint)...")
server_process = start_server()
response = requests.get(f"{base_url}/products")
assert response.status_code == 200
assert isinstance(response.json(), list)
assert len(response.json()) >= 3
server_process.terminate()
server_process.wait()
print(" ✓ GET /products works")
# Test 2: Test GET /products with max_price query parameter
print("\n2. Testing GET /products with max_price query parameter...")
server_process = start_server()
response = requests.get(f"{base_url}/products?max_price=15")
assert response.status_code == 200
products = response.json()
assert all(p["price"] <= 15 for p in products)
server_process.terminate()
server_process.wait()
print(" ✓ GET /products with max_price works")
# Test 3: Test GET /products/{id} (existing product)
print("\n3. Testing GET /products/{id} (existing product)...")
server_process = start_server()
response = requests.get(f"{base_url}/products/1")
assert response.status_code == 200
product = response.json()
assert product["id"] == 1
assert "name" in product
assert "price" in product
assert "stock" in product
server_process.terminate()
server_process.wait()
print(" ✓ GET /products/{id} works")
# Test 4: Test GET /products/{id} (non-existing product)
print("\n4. Testing GET /products/{id} (non-existing product)...")
server_process = start_server()
response = requests.get(f"{base_url}/products/999")
assert response.status_code == 404
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ GET /products/{id} for non-existing product returns 404")
# Test 5: Test POST /login with valid credentials
print("\n5. Testing POST /login with valid credentials...")
server_process = start_server()
response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
assert response.status_code == 200
token_response = response.json()
assert "token" in token_response
assert token_response["token"]
server_process.terminate()
server_process.wait()
print(" ✓ POST /login with valid credentials works")
# Test 6: Test POST /login with invalid credentials
print("\n6. Testing POST /login with invalid credentials...")
server_process = start_server()
response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "wrongpassword"})
assert response.status_code == 401
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ POST /login with invalid credentials returns 401")
# Test 7: Test POST /login with malformed body (not JSON)
print("\n7. Testing POST /login with malformed body (not JSON)...")
server_process = start_server()
response = requests.post(f"{base_url}/login", data="not json")
assert response.status_code == 400
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ POST /login with malformed body returns 400")
# Test 8: Test POST /orders with authentication
print("\n8. Testing POST /orders with authentication...")
server_process = start_server()
# Check the initial product stock
response = requests.get(f"{base_url}/products/1")
print(f" Initial product 1 stock: {response.json()['stock']}")
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 2}, headers=headers)
print(f" Order creation response status: {response.status_code}")
print(f" Order creation response body: {response.text}")
assert response.status_code == 201
order = response.json()
assert "id" in order
assert "product_id" in order
assert "quantity" in order
assert "total" in order
assert order["product_id"] == 1
assert order["quantity"] == 2
assert order["total"] == 20.0 # 10.0 * 2
server_process.terminate()
server_process.wait()
print(" ✓ POST /orders with valid authentication works")
# Test 9: Test POST /orders without authentication
print("\n9. Testing POST /orders without authentication...")
server_process = start_server()
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 1})
assert response.status_code == 401
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ POST /orders without authentication returns 401")
# Test 10: Test POST /orders with malformed body (missing product_id)
print("\n10. Testing POST /orders with malformed body (missing product_id)...")
server_process = start_server()
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"quantity": 2}, headers=headers)
assert response.status_code == 400
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ POST /orders with malformed body returns 400")
# Test 11: Test POST /orders with invalid product_id
print("\n11. Testing POST /orders with invalid product_id...")
server_process = start_server()
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 999, "quantity": 1}, headers=headers)
assert response.status_code == 404
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ POST /orders with invalid product_id returns 404")
# Test 12: Test POST /orders with insufficient stock
print("\n12. Testing POST /orders with insufficient stock...")
server_process = start_server()
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 10}, headers=headers)
assert response.status_code == 409
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ POST /orders with insufficient stock returns 409")
# Test 13: Test GET /orders (user's orders)
print("\n13. Testing GET /orders (user's orders)...")
server_process = start_server()
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(f"{base_url}/orders", headers=headers)
assert response.status_code == 200
user_orders = response.json()
assert isinstance(user_orders, list)
# Should have 1 order (alice's order from test 8)
assert len(user_orders) == 1
assert user_orders[0]["username"] == "alice"
server_process.terminate()
server_process.wait()
print(" ✓ GET /orders returns only user's orders")
# Test 14: Test GET /orders/{id} (valid order)
print("\n14. Testing GET /orders/{id} (valid order)...")
server_process = start_server()
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
# Create an order
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 2}, headers=headers)
assert response.status_code == 201
order = response.json()
# Get the order
response = requests.get(f"{base_url}/orders/{order['id']}", headers=headers)
assert response.status_code == 200
fetched_order = response.json()
assert fetched_order["id"] == order["id"]
server_process.terminate()
server_process.wait()
print(" ✓ GET /orders/{id} works for valid order")
# Test 15: Test GET /orders/{id} (order from different user)
print("\n15. Testing GET /orders/{id} (order from different user)...")
server_process = start_server()
# Create an order with alice
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 2, "quantity": 1}, headers=headers)
assert response.status_code == 201
alice_order = response.json()
# Create an order with bob
login_response = requests.post(f"{base_url}/login", json={"username": "bob", "password": "hunter2"})
bob_token = login_response.json()["token"]
bob_headers = {"Authorization": f"Bearer {bob_token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 2, "quantity": 1}, headers=bob_headers)
assert response.status_code == 201
bob_order = response.json()
# Try to access bob's order with alice's token
response = requests.get(f"{base_url}/orders/{bob_order['id']}", headers=headers)
assert response.status_code == 404
error_response = response.json()
assert "error" in error_response or "detail" in error_response
server_process.terminate()
server_process.wait()
print(" ✓ GET /orders/{id} returns 404 for other user's order")
# Test 16: Test product stock is updated after order
print("\n16. Testing product stock is updated after order...")
server_process = start_server()
login_response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
token = login_response.json()["token"]
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 2}, headers=headers)
assert response.status_code == 201
# Check product stock
response = requests.get(f"{base_url}/products/1")
assert response.status_code == 200
product = response.json()
# Initial stock was 5, we ordered 2, so should be 3
assert product["stock"] == 3
server_process.terminate()
server_process.wait()
print(" ✓ Product stock is updated after order")
# Test 17: Test GET /products with invalid max_price parameter
print("\n17. Testing GET /products with invalid max_price parameter...")
server_process = start_server()
response = requests.get(f"{base_url}/products?max_price=invalid")
assert response.status_code == 200
products = response.json()
# Should return all products when invalid max_price
assert len(products) >= 3
server_process.terminate()
server_process.wait()
print(" ✓ GET /products with invalid max_price returns all products")
print("\n=== ALL TESTS PASSED ===")
if __name__ == "__main__":
test_api()
fastapi uvicorn
#!/usr/bin/env python3
import subprocess
import time
import requests
import sys
import json
# Start the server
print("Starting the server...")
server_process = subprocess.Popen(
["python3", "app.py"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# Wait for server to start
time.sleep(3)
# Test endpoints
base_url = "http://127.0.0.1:8000"
print("\n1. Testing GET /products (public endpoint)...")
response = requests.get(f"{base_url}/products")
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n2. Testing GET /products with max_price query parameter...")
response = requests.get(f"{base_url}/products?max_price=15")
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n3. Testing GET /products/{id} (existing product)...")
response = requests.get(f"{base_url}/products/1")
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n4. Testing GET /products/{id} (non-existing product)...")
response = requests.get(f"{base_url}/products/999")
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n5. Testing POST /login with valid credentials (alice/password123)...")
response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "password123"})
print(f" Status code: {response.status_code}")
if response.status_code == 200:
token = response.json()["token"]
print(f" Token: {token}")
else:
print(f" Response: {response.json()}")
print("\n6. Testing POST /orders (valid order)...")
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 2}, headers=headers)
print(f" Status code: {response.status_code}")
if response.status_code == 201:
order = response.json()
print(f" Order created: {order}")
else:
print(f" Response: {response.json()}")
print("\n7. Testing GET /orders (user's orders)...")
response = requests.get(f"{base_url}/orders", headers=headers)
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n8. Testing GET /orders/{id} (valid order)...")
response = requests.get(f"{base_url}/orders/{order['id']}", headers=headers)
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n9. Testing POST /login with invalid credentials...")
response = requests.post(f"{base_url}/login", json={"username": "alice", "password": "wrongpassword"})
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
print("\n10. Testing POST /orders without authentication...")
response = requests.post(f"{base_url}/orders", json={"product_id": 1, "quantity": 1})
print(f" Status code: {response.status_code}")
print(f" Response: {response.json()}")
# Clean up
server_process.terminate()
server_process.wait()
print("\nServer stopped.").........F.F.......F..... [100%]
=================================== FAILURES ===================================
_________________________ test_create_order_and_total __________________________
base_url = 'http://127.0.0.1:34739'
alice_token = 'jjY3R8AokgeRIidlj0FeBTXrcJHTr0Ubuy60GQQ3WNE'
def test_create_order_and_total(base_url, alice_token):
# discover product 1's price to validate total
_, prod, _ = _request("GET", f"{base_url}/products/1")
price = prod["price"]
status, order, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1, "quantity": 3})
> assert status == 201, f"POST /orders -> {status} ({raw})"
E AssertionError: POST /orders -> 200 ({"id":1,"user_id":"alice","product_id":1,"quantity":3,"total":30.0})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:223: AssertionError
___________________________ test_orders_are_per_user ___________________________
base_url = 'http://127.0.0.1:34739'
alice_token = 'jjY3R8AokgeRIidlj0FeBTXrcJHTr0Ubuy60GQQ3WNE'
def test_orders_are_per_user(base_url, alice_token):
# alice creates an order
status, _, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 2, "quantity": 1})
> assert status == 201, f"alice POST /orders -> {status} ({raw})"
E AssertionError: alice POST /orders -> 200 ({"id":2,"user_id":"alice","product_id":2,"quantity":1,"total":20.5})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:240: AssertionError
_________________________ test_order_decrements_stock __________________________
base_url = 'http://127.0.0.1:34739'
alice_token = 'jjY3R8AokgeRIidlj0FeBTXrcJHTr0Ubuy60GQQ3WNE'
def test_order_decrements_stock(base_url, alice_token):
_, before, _ = _request("GET", f"{base_url}/products/3")
start = before["stock"]
status, _, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 3, "quantity": 2})
> assert status == 201, f"POST /orders -> {status} ({raw})"
E AssertionError: POST /orders -> 200 ({"id":3,"user_id":"alice","product_id":3,"quantity":2,"total":31.98})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:307: AssertionError
=========================== short test summary info ============================
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_create_order_and_total
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_orders_are_per_user
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_decrements_stock
3 failed, 22 passed in 0.62s
score: 22/25 (88%)
import os
import secrets
from typing import Dict, List, Optional
from fastapi import FastAPI, Request, HTTPException as FastAPIHTTPException, Depends, Header
from fastapi.responses import JSONResponse
import uvicorn
# In-memory data storage
users: Dict[str, str] = {"alice": "password123", "bob": "hunter2"}
products: List[Dict] = [
{"id": 1, "name": "Product 1", "price": 10.0, "stock": 5},
{"id": 2, "name": "Product 2", "price": 20.5, "stock": 7},
{"id": 3, "name": "Product 3", "price": 15.99, "stock": 10},
]
orders: List[Dict] = []
tokens: Dict[str, str] = {} # token -> username
order_id_counter = 1
# Helper functions
def generate_token() -> str:
return secrets.token_urlsafe(32)
def validate_token(token: str) -> Optional[str]:
return tokens.get(token)
def error_response(message: str, status_code: int) -> JSONResponse:
return JSONResponse(
status_code=status_code,
content={"error": message},
headers={"Content-Type": "application/json"},
)
# FastAPI app
app = FastAPI()
# Exception handler for HTTPException to format error responses
@app.exception_handler(FastAPIHTTPException)
async def http_exception_handler(request: Request, exc: FastAPIHTTPException):
# exc.detail can be dict, string, or list
if isinstance(exc.detail, dict) and "error" in exc.detail:
# Already formatted error dict
return JSONResponse(
status_code=exc.status_code,
content=exc.detail,
headers={"Content-Type": "application/json"},
)
# If detail is a string, wrap it in {"error": detail}
if isinstance(exc.detail, str):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
headers={"Content-Type": "application/json"},
)
# Default: convert to string and wrap
return JSONResponse(
status_code=exc.status_code,
content={"error": str(exc.detail)},
headers={"Content-Type": "application/json"},
)
# Public endpoints
@app.get("/products")
async def list_products(max_price: Optional[str] = None) -> List[Dict]:
filtered = products
if max_price is not None:
try:
max_price_val = float(max_price)
filtered = [p for p in products if p["price"] <= max_price_val]
except ValueError:
pass # ignore invalid max_price, return all
return filtered
@app.get("/products/{product_id}")
async def get_product(product_id: int) -> Dict:
for product in products:
if product["id"] == product_id:
return product
raise FastAPIHTTPException(status_code=404, detail={"error": "Product not found"})
# Authentication
@app.post("/login")
async def login(request: Request) -> Dict:
try:
body = await request.json()
except:
raise FastAPIHTTPException(status_code=400, detail={"error": "Invalid JSON"})
username = body.get("username")
password = body.get("password")
if not username or not password:
raise FastAPIHTTPException(status_code=400, detail={"error": "Missing username or password"})
if users.get(username) != password:
raise FastAPIHTTPException(status_code=401, detail={"error": "Invalid credentials"})
token = generate_token()
tokens[token] = username
return {"token": token}
# Dependency for protected endpoints
async def get_current_user(authorization: Optional[str] = Header(None)) -> str:
if not authorization or not authorization.startswith("Bearer "):
raise FastAPIHTTPException(status_code=401, detail={"error": "Missing or invalid token"})
token = authorization.split(" ", 1)[1]
username = validate_token(token)
if not username:
raise FastAPIHTTPException(status_code=401, detail={"error": "Invalid token"})
return username
# Protected endpoints
@app.post("/orders")
async def create_order(
request: Request,
current_user: str = Depends(get_current_user)
) -> Dict:
try:
body = await request.json()
except:
raise FastAPIHTTPException(status_code=400, detail={"error": "Invalid JSON"})
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
raise FastAPIHTTPException(status_code=400, detail={"error": "Missing product_id or quantity"})
if not isinstance(product_id, int) or not isinstance(quantity, int):
raise FastAPIHTTPException(status_code=400, detail={"error": "product_id and quantity must be integers"})
if quantity < 1:
raise FastAPIHTTPException(status_code=400, detail={"error": "quantity must be at least 1"})
# Find product
product = None
for p in products:
if p["id"] == product_id:
product = p
break
if not product:
raise FastAPIHTTPException(status_code=404, detail={"error": "Product not found"})
if quantity > product["stock"]:
raise FastAPIHTTPException(status_code=409, detail={"error": "Insufficient stock"})
# Create order
global order_id_counter
order = {
"id": order_id_counter,
"user_id": current_user,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
}
orders.append(order)
order_id_counter += 1
# Reduce stock
product["stock"] -= quantity
return order
@app.get("/orders")
async def list_orders(current_user: str = Depends(get_current_user)) -> List[Dict]:
user_orders = [o for o in orders if o["user_id"] == current_user]
return user_orders
@app.get("/orders/{order_id}")
async def get_order(order_id: int, current_user: str = Depends(get_current_user)) -> Dict:
for order in orders:
if order["id"] == order_id:
if order["user_id"] != current_user:
raise FastAPIHTTPException(status_code=404, detail={"error": "Order not found"})
return order
raise FastAPIHTTPException(status_code=404, detail={"error": "Order not found"})
# Run the app
if __name__ == "__main__":
host = os.getenv("HOST", "127.0.0.1")
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi>=0.100 uvicorn>=0.20
INFO: Started server process [132] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: 127.0.0.1:35238 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:35252 - "POST /orders HTTP/1.1" 401 Unauthorized INFO: 127.0.0.1:35254 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:35096 - "POST /orders HTTP/1.1" 200 OK INFO: 127.0.0.1:35110 - "GET /products/1 HTTP/1.1" 200 OK INFO: 127.0.0.1:35116 - "POST /orders HTTP/1.1" 409 Conflict INFO: 127.0.0.1:35124 - "GET /products/1 HTTP/1.1" 200 OK INFO: 127.0.0.1:35134 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:41118 - "GET /orders HTTP/1.1" 401 Unauthorized INFO: 127.0.0.1:41128 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:41132 - "GET /orders HTTP/1.1" 200 OK INFO: 127.0.0.1:51796 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:51812 - "GET /orders HTTP/1.1" 200 OK INFO: 127.0.0.1:51818 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:51830 - "POST /orders HTTP/1.1" 401 Unauthorized INFO: 127.0.0.1:51474 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:51478 - "POST /orders HTTP/1.1" 200 OK INFO: 127.0.0.1:51488 - "GET /orders HTTP/1.1" 200 OK INFO: 127.0.0.1:51504 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:51508 - "GET /orders/1 HTTP/1.1" 404 Not Found INFO: 127.0.0.1:60220 - "GET /orders/1 HTTP/1.1" 200 OK INFO: 127.0.0.1:60226 - "GET /orders/999 HTTP/1.1" 401 Unauthorized INFO: 127.0.0.1:60238 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:60250 - "GET /orders/999 HTTP/1.1" 404 Not Found INFO: 127.0.0.1:46752 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:46766 - "POST /orders HTTP/1.1" 400 Bad Request INFO: 127.0.0.1:46782 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:46790 - "POST /orders HTTP/1.1" 400 Bad Request INFO: 127.0.0.1:46792 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:46796 - "POST /orders HTTP/1.1" 400 Bad Request INFO: 127.0.0.1:51216 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:51226 - "POST /orders HTTP/1.1" 400 Bad Request INFO: 127.0.0.1:51228 - "POST /login HTTP/1.1" 200 OK INFO: 127.0.0.1:51240 - "GET /products/2 HTTP/1.1" 200 OK INFO: 127.0.0.1:51242 - "POST /orders HTTP/1.1" 200 OK INFO: 127.0.0.1:51250 - "GET /products/2 HTTP/1.1" 200 OK INFO: 127.0.0.1:44780 - "GET /products/2 HTTP/1.1" 200 OK INFO: 127.0.0.1:44790 - "POST /login HTTP/1.1" 401 Unauthorized INFO: Shutting down INFO: Waiting for application shutdown. INFO: Application shutdown complete. INFO: Finished server process [132]
INFO: Started server process [213] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8001 (Press CTRL+C to quit) INFO: 172.23.0.5:51712 - "GET /v1/models HTTP/1.1" 404 Not Found INFO: 127.0.0.1:53286 - "GET /products HTTP/1.1" 200 OK INFO: Shutting down INFO: Waiting for application shutdown. INFO: Application shutdown complete. INFO: Finished server process [213]
INFO: Started server process [224] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: 127.0.0.1:47626 - "GET /products HTTP/1.1" 200 OK INFO: Shutting down INFO: Waiting for application shutdown. INFO: Application shutdown complete. INFO: Finished server process [224]
......................... [100%] 25 passed in 0.65s score: 25/25 (100%)
import os
import json
import base64
import secrets
from typing import Dict, List, Optional
from fastapi import FastAPI, Request, Response, HTTPException, Header, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel, validator
# Configuration
HOST = os.getenv("HOST", "127.0.0.1")
PORT = int(os.getenv("PORT", "8000"))
# In-memory data store
users = {
"alice": "password123",
"bob": "hunter2",
}
products = [
{"id": 1, "name": "Laptop", "price": 999.99, "stock": 5},
{"id": 2, "name": "Mouse", "price": 25.50, "stock": 10},
{"id": 3, "name": "Keyboard", "price": 45.00, "stock": 8},
]
orders = []
next_order_id = 1
token_to_username: Dict[str, str] = {}
# Pydantic models
class LoginRequest(BaseModel):
username: str
password: str
class CreateOrderRequest(BaseModel):
product_id: int
quantity: int
@validator("quantity")
def positive_quantity(cls, v):
if v < 1:
raise ValueError("quantity must be at least 1")
return v
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class Config:
from_attributes = True
class Order(BaseModel):
id: int
user_id: str
product_id: int
quantity: int
total: float
class Config:
from_attributes = True
app = FastAPI()
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
# Return the detail as is if it's a dict, otherwise wrap in error field
if isinstance(exc.detail, dict):
content = exc.detail
else:
content = {"error": str(exc.detail)}
return JSONResponse(status_code=exc.status_code, content=content)
# Helper functions
def generate_token() -> str:
# Generate a URL-safe random token
return base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("utf-8")
def authenticate_user(username: str, password: str) -> Optional[str]:
if username in users and users[username] == password:
return username
return None
def get_user_from_token(authorization: Optional[str] = None) -> str:
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail={"error": "Missing or invalid token"})
token = authorization.split("Bearer ", 1)[1]
username = token_to_username.get(token)
if not username:
raise HTTPException(status_code=401, detail={"error": "Invalid token"})
return username
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON"})
# Validate body has required fields
if not isinstance(body, dict) or "username" not in body or "password" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing username or password"})
username = body["username"]
password = body["password"]
authenticated_user = authenticate_user(username, password)
if not authenticated_user:
raise HTTPException(status_code=401, detail={"error": "Invalid credentials"})
token = generate_token()
token_to_username[token] = authenticated_user
return JSONResponse(content={"token": token})
@app.get("/products")
async def get_products(max_price: Optional[str] = Query(None, alias="max_price")):
filtered = products
if max_price is not None:
try:
max_price_float = float(max_price)
filtered = [p for p in products if p["price"] <= max_price_float]
except ValueError:
pass
return JSONResponse(content=filtered)
@app.get("/products/{id}")
async def get_product(id: int):
for p in products:
if p["id"] == id:
return JSONResponse(content=p)
raise HTTPException(status_code=404, detail={"error": "Product not found"})
@app.post("/orders")
async def create_order(request: Request, authorization: Optional[str] = Header(None)):
username = get_user_from_token(authorization)
try:
body = await request.json()
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON"})
# Validate using Pydantic model
try:
order_data = CreateOrderRequest(**body)
except Exception as e:
# ValidationError and other errors
raise HTTPException(status_code=400, detail={"error": "Invalid request data"})
product_id = order_data.product_id
quantity = order_data.quantity
# Find product
product = next((p for p in products if p["id"] == product_id), None)
if not product:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
if quantity > product["stock"]:
raise HTTPException(status_code=409, detail={"error": "Insufficient stock"})
# Create order
global next_order_id
order = {
"id": next_order_id,
"user_id": username,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
}
orders.append(order)
next_order_id += 1
# Update product stock
product["stock"] -= quantity
return JSONResponse(status_code=201, content=order)
@app.get("/orders")
async def get_orders(request: Request, authorization: Optional[str] = Header(None)):
username = get_user_from_token(authorization)
user_orders = [o for o in orders if o["user_id"] == username]
return JSONResponse(content=user_orders)
@app.get("/orders/{id}")
async def get_order(id: int, request: Request, authorization: Optional[str] = Header(None)):
username = get_user_from_token(authorization)
for order in orders:
if order["id"] == id and order["user_id"] == username:
return JSONResponse(content=order)
raise HTTPException(status_code=404, detail={"error": "Order not found"})
def run_app():
import uvicorn
uvicorn.run(app, host=HOST, port=PORT)
if __name__ == "__main__":
run_app()
fastapi uvicorn
EFFF.....EEEFFEEEE.EEEEE. [100%]
==================================== ERRORS ====================================
_____________________ ERROR at setup of test_login_success _____________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
________________ ERROR at setup of test_create_order_and_total _________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
_____________ ERROR at setup of test_create_order_missing_product ______________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
__________________ ERROR at setup of test_orders_are_per_user __________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
_________________ ERROR at setup of test_order_malformed_json __________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
_________________ ERROR at setup of test_order_missing_fields __________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
_______________ ERROR at setup of test_order_non_integer_fields ________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
____________ ERROR at setup of test_order_zero_or_negative_quantity ____________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
________________ ERROR at setup of test_order_decrements_stock _________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
____________ ERROR at setup of test_order_exceeding_stock_conflict _____________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
_____________________ ERROR at setup of test_get_own_order _____________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
___________________ ERROR at setup of test_get_missing_order ___________________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
_____________ ERROR at setup of test_get_other_users_order_is_404 ______________
base_url = 'http://127.0.0.1:41329'
@pytest.fixture(scope="session")
def alice_token(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "password123"})
> assert status == 200, f"login(alice) -> {status}, body={raw}"
E AssertionError: login(alice) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:147: AssertionError
=================================== FAILURES ===================================
____________________________ test_login_second_user ____________________________
base_url = 'http://127.0.0.1:41329'
def test_login_second_user(base_url):
status, body, raw = _request("POST", f"{base_url}/login",
body={"username": "bob", "password": "hunter2"})
> assert status == 200, f"login(bob) -> {status}, body={raw}"
E AssertionError: login(bob) -> 422, body={"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]}
E assert 422 == 200
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:162: AssertionError
___________________________ test_login_bad_password ____________________________
base_url = 'http://127.0.0.1:41329'
def test_login_bad_password(base_url):
status, _, raw = _request("POST", f"{base_url}/login",
body={"username": "alice", "password": "wrong"})
> assert status == 401, f"bad password should be 401, got {status} ({raw})"
E AssertionError: bad password should be 401, got 422 ({"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]})
E assert 422 == 401
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:169: AssertionError
___________________________ test_login_unknown_user ____________________________
base_url = 'http://127.0.0.1:41329'
def test_login_unknown_user(base_url):
status, _, raw = _request("POST", f"{base_url}/login",
body={"username": "nobody", "password": "x"})
> assert status == 401, f"unknown user should be 401, got {status} ({raw})"
E AssertionError: unknown user should be 401, got 422 ({"detail":[{"type":"missing","loc":["query","username"],"msg":"Field required","input":null},{"type":"missing","loc":["query","password"],"msg":"Field required","input":null}]})
E assert 422 == 401
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:175: AssertionError
__________________________ test_login_malformed_json ___________________________
base_url = 'http://127.0.0.1:41329'
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': [{'input': None, 'loc': ['query', 'username'], 'msg': 'Field required', 'type': 'missing'}, {'input': None, 'loc': ['query', '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': ['query', 'username'], 'msg': 'Field required', 'input': None}, {'type': 'missing', 'loc': ['query', 'password'], 'msg': 'Field required', 'input': None}]})
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:41329'
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': None, 'loc': ['query', 'username'], 'msg': 'Field required', 'type': 'missing'}, {'input': None, 'loc': ['query', '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': ['query', 'username'], 'msg': 'Field required', 'input': None}, {'type': 'missing', 'loc': ['query', 'password'], 'msg': 'Field required', 'input': None}]})
E assert 422 == 400
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:89: AssertionError
=========================== short test summary info ============================
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_second_user
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_bad_password
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_unknown_user
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
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_success
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_create_order_and_total
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_create_order_missing_product
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_orders_are_per_user
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_malformed_json
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_missing_fields
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_non_integer_fields
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_zero_or_negative_quantity
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_decrements_stock
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_exceeding_stock_conflict
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_own_order
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_missing_order
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_other_users_order_is_404
5 failed, 7 passed, 13 errors in 0.73s
score: 7/25 (28%)
import os
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from fastapi import FastAPI, HTTPException, Depends, Query, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import JSONResponse
from pydantic import BaseModel, validator
app = FastAPI()
# Security scheme for Bearer tokens
security = HTTPBearer()
# In-memory data stores
users: Dict[str, str] = {
"alice": "password123",
"bob": "hunter2"
}
products: Dict[int, Dict] = {}
orders: Dict[int, Dict] = {}
next_order_id = 1
# Initialize with seed data
products[1] = {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10}
products[2] = {"id": 2, "name": "Mouse", "price": 29.99, "stock": 20}
products[3] = {"id": 3, "name": "Keyboard", "price": 79.99, "stock": 15}
# User-specific order storage
user_orders: Dict[str, List[int]] = {}
def generate_token(username: str) -> str:
"""Generate a simple token based on username for demo purposes"""
return f"token_{username}_{os.urandom(4).hex()}"
def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str:
"""Get current user from token (simple implementation)"""
# For demo, we'll use a simple token format and map it back to users
token = credentials.credentials
# Extract username from token (simple parsing for demo)
if token.startswith("token_"):
parts = token.split("_")
if len(parts) >= 3:
username = parts[1]
if username in users:
return username
raise HTTPException(status_code=401, detail={"error": "Invalid or expired token"})
@app.post("/login")
async def login(username: str, password: str):
"""Login endpoint to authenticate users and return token"""
if username not in users or users[username] != password:
raise HTTPException(status_code=401, detail={"error": "Invalid credentials"})
token = generate_token(username)
return {"token": token}
@app.get("/products")
async def get_products(max_price: Optional[str] = Query(None, description="Maximum price filter")):
"""Get all products, optionally filtered by max_price"""
try:
filtered_products = list(products.values())
if max_price is not None:
try:
max_price_float = float(max_price)
filtered_products = [p for p in filtered_products if p["price"] <= max_price_float]
except ValueError:
# Invalid max_price format, return all products
pass
return filtered_products
except Exception as e:
raise HTTPException(status_code=500, detail={"error": "Internal server error"})
@app.get("/products/{product_id}")
async def get_product(product_id: int):
"""Get a specific product by ID"""
if product_id not in products:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
return products[product_id]
class OrderCreate(BaseModel):
"""Schema for creating an order"""
product_id: int
quantity: int
@validator("quantity")
def validate_quantity(cls, v):
if v < 1:
raise ValueError("Quantity must be at least 1")
return v
@app.post("/orders")
async def create_order(order_data: OrderCreate, current_user: str = Depends(get_current_user)):
"""Create a new order for the authenticated user"""
product_id = order_data.product_id
quantity = order_data.quantity
# Check if product exists
if product_id not in products:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
product = products[product_id]
# Check if enough stock
if quantity > product["stock"]:
raise HTTPException(status_code=409, detail={"error": "Not enough stock available"})
# Calculate total
total = product["price"] * quantity
# Create order
global next_order_id
order_id = next_order_id
next_order_id += 1
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": current_user # Store the username for ownership validation
}
orders[order_id] = order
# Add to user's order list
if current_user not in user_orders:
user_orders[current_user] = []
user_orders[current_user].append(order_id)
# Reduce product stock
product["stock"] -= quantity
return order
@app.get("/orders")
async def get_orders(current_user: str = Depends(get_current_user)):
"""Get all orders for the authenticated user"""
if current_user not in user_orders:
return []
user_order_ids = user_orders[current_user]
user_orders_list = [orders[oid] for oid in user_order_ids if oid in orders]
# Remove username from response
for order in user_orders_list:
if "username" in order:
del order["username"]
return user_orders_list
@app.get("/orders/{order_id}")
async def get_order(order_id: int, current_user: str = Depends(get_current_user)):
"""Get a specific order if it belongs to the authenticated user"""
if order_id not in orders:
raise HTTPException(status_code=404, detail={"error": "Order not found"})
order = orders[order_id]
# Check if order belongs to current user
if order.get("username") != current_user:
raise HTTPException(status_code=404, detail={"error": "Order not found"})
# Remove username from response
response_order = order.copy()
if "username" in response_order:
del response_order["username"]
return response_order
if __name__ == "__main__":
import uvicorn
host = os.getenv("HOST", "127.0.0.1")
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn[standard]
.........F.FFFFFFF.FF.FF. [100%]
=================================== FAILURES ===================================
_________________________ test_create_order_and_total __________________________
base_url = 'http://127.0.0.1:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
def test_create_order_and_total(base_url, alice_token):
# discover product 1's price to validate total
_, prod, _ = _request("GET", f"{base_url}/products/1")
price = prod["price"]
status, order, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1, "quantity": 3})
> assert status == 201, f"POST /orders -> {status} ({raw})"
E AssertionError: POST /orders -> 200 ({"id":1,"product_id":1,"quantity":3,"total":2999.9700000000003,"user":"alice"})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:223: AssertionError
___________________________ test_orders_are_per_user ___________________________
base_url = 'http://127.0.0.1:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
def test_orders_are_per_user(base_url, alice_token):
# alice creates an order
status, _, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 2, "quantity": 1})
> assert status == 201, f"alice POST /orders -> {status} ({raw})"
E AssertionError: alice POST /orders -> 200 ({"id":2,"product_id":2,"quantity":1,"total":29.99,"user":"alice"})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:240: AssertionError
__________________________ test_login_malformed_json ___________________________
base_url = 'http://127.0.0.1:60311'
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 = 400, parsed = {'detail': {'error': 'Invalid 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': 'Invalid JSON'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Invalid JSON'}}, dict)
E + and None = <built-in method get of dict object at 0x7f24ec61ec40>('error')
E + where <built-in method get of dict object at 0x7f24ec61ec40> = {'detail': {'error': 'Invalid JSON'}}.get
E + and None = <built-in method get of dict object at 0x7f24ec61ec40>('message')
E + where <built-in method get of dict object at 0x7f24ec61ec40> = {'detail': {'error': 'Invalid JSON'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
__________________________ test_login_missing_fields ___________________________
base_url = 'http://127.0.0.1:60311'
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 = 400, parsed = {'detail': {'error': 'Missing username or password'}}
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 username or password'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Missing username or password'}}, dict)
E + and None = <built-in method get of dict object at 0x7f24eb8ee640>('error')
E + where <built-in method get of dict object at 0x7f24eb8ee640> = {'detail': {'error': 'Missing username or password'}}.get
E + and None = <built-in method get of dict object at 0x7f24eb8ee640>('message')
E + where <built-in method get of dict object at 0x7f24eb8ee640> = {'detail': {'error': 'Missing username or password'}}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
__________________________ test_order_malformed_json ___________________________
base_url = 'http://127.0.0.1:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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': 'Invalid 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': 'Invalid JSON'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Invalid JSON'}}, dict)
E + and None = <built-in method get of dict object at 0x7f24ebb27d00>('error')
E + where <built-in method get of dict object at 0x7f24ebb27d00> = {'detail': {'error': 'Invalid JSON'}}.get
E + and None = <built-in method get of dict object at 0x7f24ebb27d00>('message')
E + where <built-in method get of dict object at 0x7f24ebb27d00> = {'detail': {'error': 'Invalid 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:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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 product_id or 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 product_id or quantity'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Missing product_id or quantity'}}, dict)
E + and None = <built-in method get of dict object at 0x7f24ebae2680>('error')
E + where <built-in method get of dict object at 0x7f24ebae2680> = {'detail': {'error': 'Missing product_id or quantity'}}.get
E + and None = <built-in method get of dict object at 0x7f24ebae2680>('message')
E + where <built-in method get of dict object at 0x7f24ebae2680> = {'detail': {'error': 'Missing product_id or 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:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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 and quantity must be integers'}}
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 and quantity must be integers'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'product_id and quantity must be integers'}}, dict)
E + and None = <built-in method get of dict object at 0x7f24ebb1d440>('error')
E + where <built-in method get of dict object at 0x7f24ebb1d440> = {'detail': {'error': 'product_id and quantity must be integers'}}.get
E + and None = <built-in method get of dict object at 0x7f24ebb1d440>('message')
E + where <built-in method get of dict object at 0x7f24ebb1d440> = {'detail': {'error': 'product_id and quantity must be integers'}}.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:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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 0x7f24ebacf8c0>('error')
E + where <built-in method get of dict object at 0x7f24ebacf8c0> = {'detail': {'error': 'Quantity must be at least 1'}}.get
E + and None = <built-in method get of dict object at 0x7f24ebacf8c0>('message')
E + where <built-in method get of dict object at 0x7f24ebacf8c0> = {'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_decrements_stock __________________________
base_url = 'http://127.0.0.1:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
def test_order_decrements_stock(base_url, alice_token):
_, before, _ = _request("GET", f"{base_url}/products/3")
start = before["stock"]
status, _, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 3, "quantity": 2})
> assert status == 201, f"POST /orders -> {status} ({raw})"
E AssertionError: POST /orders -> 200 ({"id":3,"product_id":3,"quantity":2,"total":159.98,"user":"alice"})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:307: AssertionError
_____________________ test_order_exceeding_stock_conflict ______________________
base_url = 'http://127.0.0.1:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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': 'Not enough stock available'}}
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': 'Not enough stock available'}}
E assert (True and (None or None))
E + where True = isinstance({'detail': {'error': 'Not enough stock available'}}, dict)
E + and None = <built-in method get of dict object at 0x7f24eba63c80>('error')
E + where <built-in method get of dict object at 0x7f24eba63c80> = {'detail': {'error': 'Not enough stock available'}}.get
E + and None = <built-in method get of dict object at 0x7f24eba63c80>('message')
E + where <built-in method get of dict object at 0x7f24eba63c80> = {'detail': {'error': 'Not enough stock available'}}.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:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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 0x7f24ebabc400>('error')
E + where <built-in method get of dict object at 0x7f24ebabc400> = {'detail': {'error': 'Order not found'}}.get
E + and None = <built-in method get of dict object at 0x7f24ebabc400>('message')
E + where <built-in method get of dict object at 0x7f24ebabc400> = {'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:60311'
alice_token = 'aaecc456f9cac621554f6438f4a07b03'
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 0x7f24ebad2a40>('error')
E + where <built-in method get of dict object at 0x7f24ebad2a40> = {'detail': {'error': 'Order not found'}}.get
E + and None = <built-in method get of dict object at 0x7f24ebad2a40>('message')
E + where <built-in method get of dict object at 0x7f24ebad2a40> = {'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_create_order_and_total
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_orders_are_per_user
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_decrements_stock
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
12 failed, 13 passed in 0.73s
score: 13/25 (52%)
import os
from typing import List, Dict, Optional
from fastapi import FastAPI, HTTPException, Request, Header, Depends
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, ValidationError
import secrets
import json
app = FastAPI()
# Security scheme for Bearer token
security = HTTPBearer()
# In-memory data structures
users_db = {
"alice": "password123",
"bob": "hunter2"
}
# Seed products with id 1, 2, 3
products_db = {
1: {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
2: {"id": 2, "name": "Mouse", "price": 29.99, "stock": 15},
3: {"id": 3, "name": "Keyboard", "price": 79.99, "stock": 20}
}
# In-memory storage for orders and their users
orders_db = {}
orders_counter = 1
user_orders = {}
# Simple token storage - in real apps you'd use JWT or similar
token_to_user = {}
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
"""Get all products, optionally filtered by max_price"""
try:
# Convert max_price to float if provided and valid
if max_price is not None:
max_price_float = float(max_price)
filtered_products = [
product for product in products_db.values()
if product["price"] <= max_price_float
]
return filtered_products
else:
return list(products_db.values())
except ValueError:
# If max_price is not a valid number, return all products
return list(products_db.values())
@app.get("/products/{product_id}")
async def get_product(product_id: int):
"""Get a specific product by ID"""
if product_id in products_db:
return products_db[product_id]
else:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
@app.post("/login")
async def login(request: Request):
"""Authenticate user and return a token"""
try:
body = await request.json()
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON"})
if not isinstance(body, dict) or "username" not in body or "password" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing username or password"})
username = body["username"]
password = body["password"]
if username in users_db and users_db[username] == password:
# Generate a simple token
token = secrets.token_hex(16)
token_to_user[token] = username
return {"token": token}
else:
raise HTTPException(status_code=401, detail={"error": "Invalid credentials"})
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Verify Bearer token and return the authenticated user"""
token = credentials.credentials
if token in token_to_user:
return token_to_user[token]
else:
raise HTTPException(status_code=401, detail={"error": "Invalid token"})
@app.post("/orders")
async def create_order(
request: Request,
current_user: str = Depends(get_current_user)
):
"""Create a new order for the authenticated user"""
try:
body = await request.json()
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON"})
# Validate body structure
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail={"error": "Invalid order data"})
if "product_id" not in body or "quantity" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing product_id or quantity"})
try:
product_id = int(body["product_id"])
quantity = int(body["quantity"])
except (ValueError, TypeError):
raise HTTPException(status_code=400, detail={"error": "product_id and quantity must be integers"})
if quantity < 1:
raise HTTPException(status_code=400, detail={"error": "Quantity must be at least 1"})
# Check if product exists
if product_id not in products_db:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
product = products_db[product_id]
if quantity > product["stock"]:
raise HTTPException(status_code=409, detail={"error": "Not enough stock available"})
# Create order
global orders_counter
order_id = orders_counter
orders_counter += 1
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"user": current_user
}
# Store order and track which user created it
orders_db[order_id] = order
if current_user not in user_orders:
user_orders[current_user] = []
user_orders[current_user].append(order_id)
# Update product stock
products_db[product_id]["stock"] -= quantity
return order
@app.get("/orders")
async def get_user_orders(current_user: str = Depends(get_current_user)):
"""Get all orders for the authenticated user"""
if current_user not in user_orders:
return []
user_order_ids = user_orders[current_user]
user_orders_list = []
for order_id in user_order_ids:
if order_id in orders_db:
user_orders_list.append(orders_db[order_id])
return user_orders_list
@app.get("/orders/{order_id}")
async def get_user_order(
order_id: int,
current_user: str = Depends(get_current_user)
):
"""Get a specific order for the authenticated user"""
if order_id not in orders_db:
raise HTTPException(status_code=404, detail={"error": "Order not found"})
order = orders_db[order_id]
if order["user"] != current_user:
raise HTTPException(status_code=404, detail={"error": "Order not found"})
return order
if __name__ == "__main__":
import uvicorn
host = os.getenv("HOST", "127.0.0.1")
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn