import os
import uuid
from fastapi import FastAPI, Header, Query, Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
app = FastAPI()
# Convert FastAPI's 422 (validation error) → 400 so that invalid JSON bodies
# in the login endpoint don't leak a 422 status code.
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(status_code=400, content={"error": "Invalid request body."})
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
USERS: dict[str, dict] = {
"alice": {"id": 1, "password": "password123"},
"bob": {"id": 2, "password": "hunter2"},
}
PRODUCTS: dict[int, dict] = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 20},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 89.50, "stock": 15},
3: {"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 10},
}
_orders_next_id: int = 1
ORDERS: list[dict] = []
TOKENS: dict[str, str] = {}
def _auth_user(authorization: str | None) -> tuple[str, int]:
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail={"error": "Missing or malformed Authorization header."})
token = authorization[len("Bearer "):]
username = TOKENS.get(token)
if not username:
raise HTTPException(status_code=401, detail={"error": "Invalid token."})
return username, USERS[username]["id"]
def _order_view(o: dict) -> dict:
return {"id": o["id"], "product_id": o["product_id"],
"quantity": o["quantity"], "total": o["total"]}
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.get("/products")
def list_products(max_price: str | None = Query(None)):
result = list(PRODUCTS.values())
if max_price is not None:
try:
cutoff = float(max_price)
except (ValueError, TypeError):
pass
else:
result = [p for p in result if p["price"] <= cutoff]
return result
@app.get("/products/{product_id}")
def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if not product:
raise HTTPException(status_code=404, detail={"error": "Product not found."})
return product
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
@app.post("/login")
async def login(request: Request):
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Body must be valid JSON."})
if not isinstance(data, dict):
raise HTTPException(status_code=400, detail={"error": "Body must be a JSON object."})
username = data.get("username")
password = data.get("password")
if not username or not password:
raise HTTPException(status_code=400, detail={"error": "Missing username or password."})
user = USERS.get(username)
if not user or user["password"] != password:
raise HTTPException(status_code=401, detail={"error": "Invalid credentials."})
token = uuid.uuid4().hex
TOKENS[token] = username
return {"token": token}
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
@app.post("/orders", status_code=201)
async def create_order(request: Request, authorization: str | None = Header(None)):
username, user_id = _auth_user(authorization)
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Body must be valid JSON."})
if not isinstance(data, dict):
raise HTTPException(status_code=400, detail={"error": "Body must be a JSON object."})
if "product_id" not in data or "quantity" not in data:
raise HTTPException(status_code=400, detail={"error": "Missing product_id or quantity."})
product_id = data["product_id"]
quantity = data["quantity"]
# Must be ints (reject booleans and floats)
if isinstance(product_id, bool) or not isinstance(product_id, int):
raise HTTPException(status_code=400, detail={"error": "product_id must be an integer."})
if isinstance(quantity, bool) or not isinstance(quantity, int):
raise HTTPException(status_code=400, detail={"error": "quantity must be an integer."})
if quantity < 1:
raise HTTPException(status_code=400, detail={"error": "quantity must be at least 1."})
product = PRODUCTS.get(product_id)
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."})
product["stock"] -= quantity
global _orders_next_id
order_id = _orders_next_id
_orders_next_id += 1
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
"user_id": user_id,
}
ORDERS.append(order)
return _order_view(order)
@app.get("/orders")
def list_orders(authorization: str | None = Header(None)):
_username, user_id = _auth_user(authorization)
return [_order_view(o) for o in ORDERS if o["user_id"] == user_id]
@app.get("/orders/{order_id}")
def get_order(order_id: int, authorization: str | None = Header(None)):
_username, user_id = _auth_user(authorization)
for o in ORDERS:
if o["id"] == order_id and o["user_id"] == user_id:
return _order_view(o)
raise HTTPException(status_code=404, detail={"error": "Order not found."})
# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run("app:app", host=host, port=port, reload=False)
#!/usr/bin/env bash exec python3 "$@"
fastapi uvicorn
......................... [100%] 25 passed in 0.56s score: 25/25 (100%)
"""Online Store REST API — FastAPI implementation."""
import os
import secrets
from fastapi import FastAPI, Header, Query, Request
from fastapi.responses import JSONResponse
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
USERS: dict[str, str] = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS: list[dict] = [
{"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 50},
{"id": 2, "name": "Mechanical Keyboard", "price": 89.50, "stock": 30},
{"id": 3, "name": "USB-C Hub", "price": 45.0, "stock": 20},
]
ORDERS: list[dict] = []
# token -> username
TOKENS: dict[str, str] = {}
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastAPI(title="Online Store API")
def _json(status: int, body: dict) -> JSONResponse:
return JSONResponse(status_code=status, content=body)
def _unauthorized(msg: str = "Unauthorized") -> JSONResponse:
return _json(401, {"error": msg})
def _bad(msg: str = "Bad request") -> JSONResponse:
return _json(400, {"error": msg})
def _not_found(msg: str = "Not found") -> JSONResponse:
return _json(404, {"error": msg})
def _conflict(msg: str = "Conflict") -> JSONResponse:
return _json(409, {"error": msg})
def _resolve_token(authorization: str | None) -> str | None:
"""Return username if *authorization* is a valid Bearer token, else None."""
if not authorization:
return None
if not authorization.startswith("Bearer "):
return None
token = authorization[7:]
return TOKENS.get(token)
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.get("/products")
def list_products(max_price: str | None = Query(None)):
"""Return all products; optionally filter by max_price."""
products = PRODUCTS
if max_price is not None:
try:
threshold = float(max_price)
except (ValueError, TypeError):
pass
else:
products = [p for p in products if p["price"] <= threshold]
return [dict(p) for p in products]
@app.get("/products/{product_id}")
def get_product(product_id: int):
"""Return a single product by id, or 404."""
for p in PRODUCTS:
if p["id"] == product_id:
return dict(p)
return _not_found("Product not found")
# ---------------------------------------------------------------------------
# Auth endpoint
# ---------------------------------------------------------------------------
@app.post("/login")
async def login(request: Request):
"""Authenticate and return a bearer token."""
try:
body = await request.json()
except Exception:
return _bad("Request body must be valid JSON")
username = body.get("username")
password = body.get("password")
if not isinstance(username, str) or not isinstance(password, str):
return _bad("Missing or invalid 'username' or 'password'")
if USERS.get(username) != password:
return _unauthorized("Invalid credentials")
token = secrets.token_hex(32)
TOKENS[token] = username
return {"token": token}
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
def _require_user(authorization: str | None) -> str | JSONResponse:
"""Return username on success, or a JSONResponse error on failure."""
username = _resolve_token(authorization)
if username is None:
return _unauthorized("Missing or invalid token")
return username
async def _create_order_impl(
request: Request,
authorization: str | None = Header(None),
) -> JSONResponse:
user = _require_user(authorization)
if isinstance(user, JSONResponse):
return user
try:
body = await request.json()
except Exception:
return _bad("Request body must be valid JSON")
product_id = body.get("product_id")
quantity = body.get("quantity")
# --- validation ---
if product_id is None or quantity is None:
return _bad("Missing 'product_id' or 'quantity'")
if not isinstance(product_id, int) or isinstance(product_id, bool):
return _bad("'product_id' must be an integer")
if not isinstance(quantity, int) or isinstance(quantity, bool):
return _bad("'quantity' must be an integer")
if quantity < 1:
return _bad("'quantity' must be at least 1")
# --- find product ---
product = None
for p in PRODUCTS:
if p["id"] == product_id:
product = p
break
if product is None:
return _not_found("Product not found")
# --- check stock ---
if quantity > product["stock"]:
return _conflict("Insufficient stock")
# --- create order ---
order_id = len(ORDERS) + 1
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"_user": user,
}
ORDERS.append(order)
product["stock"] -= quantity
result = {k: v for k, v in order.items() if k != "_user"}
return _json(201, result)
@app.post("/orders")
async def create_order(request: Request, authorization: str | None = Header(None)):
"""Place an order (requires authentication)."""
return await _create_order_impl(request, authorization)
@app.get("/orders")
def list_orders(authorization: str | None = Header(None)):
"""Return orders placed by the authenticated user only."""
user = _require_user(authorization)
if isinstance(user, JSONResponse):
return user
return [
{k: v for k, v in o.items() if k != "_user"}
for o in ORDERS
if o.get("_user") == user
]
@app.get("/orders/{order_id}")
def get_order(order_id: int, authorization: str | None = Header(None)):
"""Return a specific order if it belongs to the authenticated user."""
user = _require_user(authorization)
if isinstance(user, JSONResponse):
return user
for o in ORDERS:
if o["id"] == order_id:
if o.get("_user") == user:
return {k: v for k, v in o.items() if k != "_user"}
return _not_found("Order not found")
return _not_found("Order not found")
# ---------------------------------------------------------------------------
# Server
# ---------------------------------------------------------------------------
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)
EEEEEEEEEEEEEEEEEEEEEEEEE [100%]
==================================== ERRORS ====================================
_____________________ ERROR at setup of test_login_success _____________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
___________________ ERROR at setup of test_login_second_user ___________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
__________________ ERROR at setup of test_login_bad_password ___________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
__________________ ERROR at setup of test_login_unknown_user ___________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_____________________ ERROR at setup of test_list_products _____________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
______________________ ERROR at setup of test_get_product ______________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
__________________ ERROR at setup of test_get_missing_product __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_________________ ERROR at setup of test_orders_requires_auth __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_______________ ERROR at setup of test_orders_rejects_bad_token ________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
________________ ERROR at setup of test_create_order_and_total _________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_____________ ERROR at setup of test_create_order_missing_product ______________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
__________________ ERROR at setup of test_orders_are_per_user __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_________________ ERROR at setup of test_login_malformed_json __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_________________ ERROR at setup of test_login_missing_fields __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_________________ ERROR at setup of test_order_malformed_json __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_________________ ERROR at setup of test_order_missing_fields __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_______________ ERROR at setup of test_order_non_integer_fields ________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
____________ ERROR at setup of test_order_zero_or_negative_quantity ____________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
__________________ ERROR at setup of test_products_have_stock __________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
________________ ERROR at setup of test_order_decrements_stock _________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
____________ ERROR at setup of test_order_exceeding_stock_conflict _____________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_____________________ ERROR at setup of test_get_own_order _____________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
___________________ ERROR at setup of test_get_missing_order ___________________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_____________ ERROR at setup of test_get_other_users_order_is_404 ______________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
_______________ ERROR at setup of test_products_max_price_filter _______________
@pytest.fixture(scope="session")
def base_url():
app_py = os.path.join(APP_DIR, "app.py")
if not os.path.isfile(app_py):
pytest.fail(f"app.py not found in {APP_DIR} (the model did not create it)")
port = _free_port()
env = dict(os.environ, HOST="127.0.0.1", PORT=str(port))
proc = subprocess.Popen(
[sys.executable, "app.py"],
cwd=APP_DIR,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + BOOT_TIMEOUT
last_err = None
while time.time() < deadline:
if proc.poll() is not None: # process exited early
out = proc.stdout.read() if proc.stdout else ""
> pytest.fail(f"app.py exited early (code {proc.returncode}). Output:\n{out}")
E Failed: app.py exited early (code 1). Output:
E Traceback (most recent call last):
E File "/tmp/bench-ws-pcqec0vb/app.py", line 5, in <module>
E from flask import Flask, request, jsonify
E ModuleNotFoundError: No module named 'flask'
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:118: Failed
=========================== short test summary info ============================
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_login_second_user
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_bad_password
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_login_unknown_user
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_list_products
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_product
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_missing_product
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_orders_requires_auth
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_orders_rejects_bad_token
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_login_malformed_json
ERROR ../../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_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_products_have_stock
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
ERROR ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_products_max_price_filter
25 errors in 0.53s
score: 0/25 (0%)
import os
import secrets
import threading
from flask import Flask, request, jsonify
# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = Flask(__name__)
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
users = {
"alice": {"username": "alice", "password": "password123"},
"bob": {"username": "bob", "password": "hunter2"},
}
products = [
{"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
{"id": 2, "name": "Mechanical Keyboard", "price": 89.99, "stock": 8},
{"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 15},
]
orders = [] # list of order dicts
order_counter = 0
order_lock = threading.Lock()
tokens = {} # token_string -> username
tokens_lock = threading.Lock()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _authenticate():
"""Return (username, error_response, status_code).
If authentication succeeds the third element is None.
"""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return None, jsonify({"error": "Missing or invalid Authorization header"}), 401
token = auth_header[7:] # strip "Bearer "
with tokens_lock:
username = tokens.get(token)
if not username:
return None, jsonify({"error": "Invalid or expired token"}), 401
return username, None, None
def _product_by_id(pid):
for p in products:
if p["id"] == pid:
return p
return None
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.route("/login", methods=["POST"])
def login():
# Parse body
try:
body = request.get_json(force=True)
except Exception:
return jsonify({"error": "Request body is not valid JSON"}), 400
if body is None or not isinstance(body, dict):
return jsonify({"error": "Request body must be a JSON object"}), 400
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return jsonify({"error": "Both 'username' and 'password' are required"}), 400
user = users.get(username)
if user is None or user["password"] != password:
return jsonify({"error": "Invalid username or password"}), 401
token = secrets.token_hex(32)
with tokens_lock:
tokens[token] = username
return jsonify({"token": token}), 200
@app.route("/products", methods=["GET"])
def get_products():
max_price_param = request.args.get("max_price")
result = list(products)
if max_price_param is not None:
try:
max_price = float(max_price_param)
except (ValueError, TypeError):
pass # not a valid number → return all
else:
result = [p for p in result if p["price"] <= max_price]
return jsonify(result), 200
@app.route("/products/<int:pid>", methods=["GET"])
def get_product(pid):
product = _product_by_id(pid)
if product is None:
return jsonify({"error": "Product not found"}), 404
return jsonify(product), 200
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
@app.route("/orders", methods=["POST"])
def create_order():
username, err, code = _authenticate()
if err:
return err, code
# Parse & validate body
try:
body = request.get_json(force=True)
except Exception:
return jsonify({"error": "Request body is not valid JSON"}), 400
if body is None or not isinstance(body, dict):
return jsonify({"error": "Request body must be a JSON object"}), 400
pid = body.get("product_id")
qty = body.get("quantity")
if pid is None or qty is None:
return jsonify({"error": "Both 'product_id' and 'quantity' are required"}), 400
# Must be integers (reject floats, strings, etc.)
if not isinstance(pid, int) or isinstance(pid, bool):
return jsonify({"error": "'product_id' must be an integer"}), 400
if not isinstance(qty, int) or isinstance(qty, bool):
return jsonify({"error": "'quantity' must be an integer"}), 400
if qty < 1:
return jsonify({"error": "'quantity' must be at least 1"}), 400
product = _product_by_id(pid)
if product is None:
return jsonify({"error": "Product not found"}), 404
with order_lock:
if product["stock"] < qty:
return jsonify({"error": "Insufficient stock"}), 409
product["stock"] -= qty
global order_counter
order_counter += 1
order_id = order_counter
order = {
"id": order_id,
"product_id": pid,
"quantity": qty,
"total": product["price"] * qty,
}
orders.append(order)
# Store owner so we can enforce per-user visibility
order["_owner"] = username
return jsonify({k: v for k, v in order.items() if k != "_owner"}), 201
@app.route("/orders", methods=["GET"])
def get_orders():
username, err, code = _authenticate()
if err:
return err, code
user_orders = [
{k: v for k, v in o.items() if k != "_owner"}
for o in orders
if o.get("_owner") == username
]
return jsonify(user_orders), 200
@app.route("/orders/<int:oid>", methods=["GET"])
def get_order(oid):
username, err, code = _authenticate()
if err:
return err, code
for o in orders:
if o["id"] == oid:
if o.get("_owner") != username:
return jsonify({"error": "Order not found"}), 404
return jsonify({k: v for k, v in o.items() if k != "_owner"}), 200
return jsonify({"error": "Order not found"}), 404
# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
app.run(host=host, port=port, debug=False)
#!/bin/sh exec /usr/bin/python3 "$@"
flask>=3.0
......................... [100%] 25 passed in 0.43s score: 25/25 (100%)
[binary file]
import json
import os
import secrets
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
# ── Seed data ────────────────────────────────────────────────────────
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = [
{"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
{"id": 2, "name": "Headphones", "price": 49.99, "stock": 20},
{"id": 3, "name": "Mouse", "price": 19.99, "stock": 50},
]
# In-memory stores
TOKENS = {} # token_string -> username
ORDERS = [] # list of order dicts
_next_order_id = 1
# ── Helpers ──────────────────────────────────────────────────────────
def json_response(handler, status, body):
"""Send a JSON response."""
payload = json.dumps(body).encode("utf-8")
handler.send_response(status)
handler.send_header("Content-Type", "application/json")
handler.send_header("Content-Length", str(len(payload)))
handler.end_headers()
handler.wfile.write(payload)
def error_response(handler, status, message):
json_response(handler, status, {"error": message})
def find_product_by_id(pid):
for p in PRODUCTS:
if p["id"] == pid:
return p
return None
def parse_body(handler):
"""Read and parse the request body as JSON. Returns (dict|None, bool)
where bool indicates success."""
try:
length = int(handler.headers.get("Content-Length", 0))
raw = handler.rfile.read(length)
body = json.loads(raw)
except (json.JSONDecodeError, ValueError, TypeError):
return None, False
if not isinstance(body, dict):
return None, False
return body, True
def authenticate(handler):
"""Return the username if the token is valid, otherwise None."""
auth = handler.headers.get("Authorization", "")
if not auth.startswith("Bearer "):
return None
token = auth[7:] # strip "Bearer "
return TOKENS.get(token)
# ── Request handler ──────────────────────────────────────────────────
class StoreHandler(BaseHTTPRequestHandler):
# Silence per-request log lines
def log_message(self, format, *args):
pass
# ── Routing ──────────────────────────────────────────────────────
def _route(self, method):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
query = parse_qs(parsed.query)
# ── Public routes ────────────────────────────────────────────
if path == "/login" and method == "POST":
return self._handle_login()
if path == "/products":
return self._handle_get_products(query)
if path.startswith("/products/"):
try:
pid = int(path.split("/products/")[1])
except (IndexError, ValueError):
return error_response(self, 404, "Not found")
return self._handle_get_product(pid)
# ── Protected routes ─────────────────────────────────────────
if path == "/orders" and method == "POST":
return self._handle_create_order()
if path == "/orders":
return self._handle_get_orders()
if path.startswith("/orders/"):
try:
oid = int(path.split("/orders/")[1])
except (IndexError, ValueError):
return error_response(self, 404, "Not found")
return self._handle_get_order(oid)
error_response(self, 404, "Not found")
# ── HTTP method dispatchers ──────────────────────────────────────
def do_GET(self):
self._route("GET")
def do_POST(self):
self._route("POST")
# ── Endpoint implementations ─────────────────────────────────────
def _handle_login(self):
body, ok = parse_body(self)
if not ok or "username" not in body or "password" not in body:
return error_response(self, 400, "Missing username or password")
username = body["username"]
password = body["password"]
if username not in USERS or USERS[username] != password:
return error_response(self, 401, "Invalid credentials")
token = secrets.token_hex(16)
TOKENS[token] = username
json_response(self, 200, {"token": token})
def _handle_get_products(self, query):
products = list(PRODUCTS)
max_price_raw = query.get("max_price", [None])[0]
if max_price_raw is not None:
try:
max_price = float(max_price_raw)
except (ValueError, TypeError):
pass # ignore invalid parameter
else:
products = [p for p in products if p["price"] <= max_price]
json_response(self, 200, products)
def _handle_get_product(self, pid):
product = find_product_by_id(pid)
if product is None:
return error_response(self, 404, "Product not found")
json_response(self, 200, product)
def _handle_create_order(self):
username = authenticate(self)
if username is None:
return error_response(self, 401, "Missing or invalid token")
body, ok = parse_body(self)
if not ok:
return error_response(self, 400, "Malformed request body")
if "product_id" not in body or "quantity" not in body:
return error_response(self, 400, "Missing product_id or quantity")
try:
product_id = int(body["product_id"])
quantity = int(body["quantity"])
except (ValueError, TypeError):
return error_response(self, 400, "product_id and quantity must be integers")
if quantity < 1:
return error_response(self, 400, "Quantity must be at least 1")
product = find_product_by_id(product_id)
if product is None:
return error_response(self, 404, "Product not found")
if quantity > product["stock"]:
return error_response(self, 409, "Insufficient stock")
global _next_order_id
total = product["price"] * quantity
order = {
"id": _next_order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
}
_next_order_id += 1
ORDERS.append(order)
# Reduce stock
product["stock"] -= quantity
response_order = {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}
json_response(self, 201, response_order)
def _handle_get_orders(self):
username = authenticate(self)
if username is None:
return error_response(self, 401, "Missing or invalid token")
user_orders = [
{
"id": o["id"],
"product_id": o["product_id"],
"quantity": o["quantity"],
"total": o["total"],
}
for o in ORDERS
if o["username"] == username
]
json_response(self, 200, user_orders)
def _handle_get_order(self, oid):
username = authenticate(self)
if username is None:
return error_response(self, 401, "Missing or invalid token")
for o in ORDERS:
if o["id"] == oid and o["username"] == username:
json_response(self, 200, {
"id": o["id"],
"product_id": o["product_id"],
"quantity": o["quantity"],
"total": o["total"],
})
return
error_response(self, 404, "Order not found")
# ── Main ─────────────────────────────────────────────────────────────
def main():
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
server = HTTPServer((host, port), StoreHandler)
print(f"Serving on http://{host}:{port}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
server.server_close()
if __name__ == "__main__":
main()
[binary file]
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
"""Online Store REST API — built with FastAPI."""
import hashlib
import os
import secrets
from fastapi import FastAPI, Header, Query, Request
from fastapi.responses import JSONResponse
# ---------------------------------------------------------------------------
# In-memory data store
# ---------------------------------------------------------------------------
USERS: dict[str, dict] = {
"alice": {"username": "alice", "password": "password123"},
"bob": {"username": "bob", "password": "hunter2"},
}
PRODUCTS: dict[int, dict] = {
1: {"id": 1, "name": "Mechanical Keyboard", "price": 79.99, "stock": 20},
2: {"id": 2, "name": "Wireless Mouse", "price": 29.99, "stock": 50},
3: {"id": 3, "name": "USB-C Hub", "price": 49.99, "stock": 15},
}
ORDERS: dict[int, dict] = {}
_next_order_id: int = 1
TOKEN_TO_USER: dict[str, str] = {}
# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def hash_password(password: str) -> str:
"""SHA-256 hash for demo purposes."""
return hashlib.sha256(password.encode()).hexdigest()
def _do_login(username: str, password: str) -> str | None:
"""Return a bearer token if credentials are valid, else None."""
user = USERS.get(username)
if user is None:
return None
if hash_password(password) != hash_password(user["password"]):
return None
token = secrets.token_hex(32)
TOKEN_TO_USER[token] = username
return token
def _resolve_token(authorization: str | None) -> str | None:
"""Parse Authorization header. Returns username or None."""
if not authorization:
return None
parts = authorization.split(" ", 1)
if len(parts) != 2 or parts[0] != "Bearer":
return None
return TOKEN_TO_USER.get(parts[1])
# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="Online Store API")
# -- Auth -------------------------------------------------------------------
@app.post("/login")
async def login_endpoint(request: Request):
try:
body = await request.json()
except Exception:
return JSONResponse(
status_code=400, content={"error": "Invalid JSON body"}
)
if not isinstance(body, dict):
return JSONResponse(
status_code=400, content={"error": "Request body must be a JSON object"}
)
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return JSONResponse(
status_code=400,
content={"error": "Missing required fields: username and password"},
)
token = _do_login(username, password)
if token is None:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
return {"token": token}
# -- Products (public) ------------------------------------------------------
@app.get("/products")
async def list_products(max_price: str | None = Query(default=None)):
result = list(PRODUCTS.values())
if max_price is not None:
try:
threshold = float(max_price)
except (ValueError, TypeError):
pass # non-numeric → return all
else:
result = [p for p in result if p["price"] <= threshold]
return result
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
return JSONResponse(status_code=404, content={"error": "Product not found"})
return product
# -- Orders (protected) -----------------------------------------------------
def _unauthorized() -> JSONResponse:
return JSONResponse(status_code=401, content={"error": "Missing or invalid token"})
def _parse_order_body(body: dict):
"""Validate order payload. Returns (product_id, quantity) or a JSONResponse."""
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
return JSONResponse(
status_code=400,
content={"error": "Missing required fields: product_id and quantity"},
)
if isinstance(product_id, bool) or not isinstance(product_id, int):
return JSONResponse(
status_code=400, content={"error": "product_id must be an integer"}
)
if isinstance(quantity, bool) or not isinstance(quantity, int):
return JSONResponse(
status_code=400, content={"error": "quantity must be an integer"}
)
if quantity < 1:
return JSONResponse(
status_code=400, content={"error": "quantity must be at least 1"}
)
return (product_id, quantity)
@app.post("/orders")
async def create_order(
request: Request,
authorization: str | None = Header(default=None),
):
username = _resolve_token(authorization)
if username is None:
return _unauthorized()
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
if not isinstance(body, dict):
return JSONResponse(
status_code=400, content={"error": "Request body must be a JSON object"}
)
result = _parse_order_body(body)
if isinstance(result, JSONResponse):
return result
product_id, quantity = result
product = PRODUCTS.get(product_id)
if product is None:
return JSONResponse(status_code=404, content={"error": "Product not found"})
if product["stock"] < quantity:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
# Deduct stock
product["stock"] -= quantity
global _next_order_id
order_id = _next_order_id
_next_order_id += 1
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
}
ORDERS[order_id] = dict(order, username=username)
return JSONResponse(content=order, status_code=201)
@app.get("/orders")
async def list_orders(authorization: str | None = Header(default=None)):
username = _resolve_token(authorization)
if username is None:
return _unauthorized()
return [
{k: v for k, v in o.items() if k != "username"}
for o in ORDERS.values()
if o["username"] == username
]
@app.get("/orders/{order_id}")
async def get_order(
order_id: int,
authorization: str | None = Header(default=None),
):
username = _resolve_token(authorization)
if username is None:
return _unauthorized()
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
return JSONResponse(status_code=404, content={"error": "Order not found"})
return {k: v for k, v in order.items() if k != "username"}
# ---------------------------------------------------------------------------
# Main entry-point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run("app:app", host=host, port=port, reload=False)
fastapi uvicorn