added trade-hub role and files

This commit is contained in:
boxswapper
2026-11-13 12:47:35 -07:00
parent 6c9c48dd19
commit 2b72f23ea9
46 changed files with 316582 additions and 0 deletions

View File

@@ -0,0 +1,31 @@
"""Test bootstrap.
Runs the suite against a throwaway in-memory SQLite database and the tiny
fixture catalog, so no PostgreSQL server is needed. Environment variables must
be set before any app module is imported, because config reads them at import
time.
"""
import os
os.environ["DATA_DIR"] = os.path.join(os.path.dirname(__file__), "fixtures")
os.environ["DATABASE_URL"] = "sqlite://"
os.environ["AUTO_INIT_DB"] = "false"
os.environ["COOKIE_SECURE"] = "false"
import pytest # noqa: E402
from app import db # noqa: E402
from app.models import Base # noqa: E402
@pytest.fixture(autouse=True)
def fresh_db():
"""Give every test an empty schema and a clean login throttle."""
engine = db.get_engine()
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
from app.auth import throttle
throttle.clear()
yield

View File

@@ -0,0 +1,30 @@
[
{
"className": "Carryable_TBO_FL_1SCU_Commodity_ProcessedGoods_ConstructionMaterials",
"name": "Construction Materials",
"type": "Cargo",
"subType": "Cargo",
"reference": "5ee15a9b-4ce5-445b-a02d-415c63f43764"
},
{
"className": "Carryable_TBO_FL_1SCU_Commodity_Metal_Iron",
"name": "Iron",
"type": "Cargo",
"subType": "Cargo",
"reference": "332096b9-1cbf-436d-b8b0-5ed422c23da9"
},
{
"className": "Carryable_TBO_FL_1SCU_Commodity_Metal_Ore_Iron",
"name": "Iron (Ore)",
"type": "Cargo",
"subType": "Cargo",
"reference": "271effd5-3aa8-4538-b9a0-b0f33dc86d68"
},
{
"className": "Carryable_TBO_FL_1SCU_Commodity_Metal_RMC",
"name": "Recycled Material Composite",
"type": "Cargo",
"subType": "Cargo",
"reference": "2e50dde5-30d4-4faa-af8f-8e5aa8004b96"
}
]

View File

@@ -0,0 +1,3 @@
[
{"className": "BEHR_Rifle_Ballistic_01", "reference": "ref-3", "itemName": "behr_rifle_ballistic_01", "type": "WeaponPersonal", "subType": "Medium", "size": 2, "grade": 1, "name": "<= PLACEHOLDER =>", "tags": "weapon", "classification": "FPS.Weapon.Medium", "stdItem": {"UUID": "fps-uuid-1", "ClassName": "BEHR_Rifle_Ballistic_01", "Size": 2, "Grade": 1, "Width": 0.15, "Height": 0.15, "Length": 0.9, "Mass": 4, "Type": "WeaponPersonal.Medium", "Name": "Behring Rifle", "Description": "A ballistic rifle.", "DescriptionText": "", "Manufacturer": {"Name": "Behring", "Code": "BEHR", "UUID": "m-behr"}, "Tags": ["weapon"], "Ports": [{"PortName": "magazine_attach", "Size": 1}]}}
]

View File

@@ -0,0 +1,6 @@
[
{"Code": "ANVL", "Name": "Anvil Aerospace", "Reference": "m1", "Description": "Anvil makes ships."},
{"Code": "BEHR", "Name": "Behring", "Reference": "m-behr", "Description": "Behring makes weapons."},
{"Code": "UNKN", "Name": "Unknown Manufacturer", "Reference": "m-unkn", "Description": ""},
{"Code": "PHLD", "Name": "<= PLACEHOLDER =>", "Reference": "m-phld", "Description": ""}
]

View File

@@ -0,0 +1,4 @@
[
{"className": "BEHR_Shield_S1", "reference": "ref-1", "itemName": "behr_shield_s1", "type": "Shield", "subType": "UNDEFINED", "size": 1, "grade": 2, "name": "<= PLACEHOLDER =>", "tags": "shield", "classification": "Ship.Shield", "stdItem": {"UUID": "si-uuid-1", "ClassName": "BEHR_Shield_S1", "Size": 1, "Grade": 2, "Width": 0.5, "Height": 0.5, "Length": 1.0, "Mass": 20, "Type": "Shield.UNDEFINED", "Name": "Behring Shield S1", "Description": "A size 1 shield.", "DescriptionText": "", "Manufacturer": {"Name": "Behring", "Code": "BEHR", "UUID": "m-behr"}, "Tags": ["shield"], "Ports": []}},
{"className": "TEST_Ph_Item", "reference": "ref-2", "itemName": "test_ph_item", "type": "PowerPlant", "subType": "UNDEFINED", "size": 2, "grade": 1, "name": "<= PLACEHOLDER =>", "tags": "", "classification": "Ship.PowerPlant", "stdItem": {"UUID": "si-uuid-2", "ClassName": "TEST_Ph_Item", "Size": 2, "Grade": 1, "Width": 0.5, "Height": 0.5, "Length": 0.5, "Mass": 10, "Type": "PowerPlant.UNDEFINED", "Name": "<= PLACEHOLDER =>", "Description": "", "DescriptionText": "", "Manufacturer": {"Name": "<= PLACEHOLDER =>", "Code": "UNKN", "UUID": "m-unkn"}, "Tags": [], "Ports": [{"PortName": "connect", "Size": 1}]}}
]

View File

@@ -0,0 +1,4 @@
[
{"UUID": "ship-uuid-1", "ClassName": "ANVL_Carrack", "Name": "Anvil Carrack", "Description": "Long-range explorer.", "DescriptionText": "Long-range explorer.", "Career": "Exploration", "Role": "Expedition", "Manufacturer": {"UUID": "m1", "Code": "ANVL", "Name": "Anvil Aerospace"}, "Size": 4, "Length": 35.0, "Width": 23.0, "Height": 8.5, "Crew": 4, "Mass": 2000000, "IsSpaceship": true, "IsVehicle": false, "IsGravlev": false},
{"UUID": "ship-uuid-2", "ClassName": "TEST_Placeholder_Ship", "Name": "<= PLACEHOLDER =>", "Description": "", "DescriptionText": "", "Career": "", "Role": "", "Manufacturer": {"UUID": "m2", "Code": "UNKN", "Name": "Unknown Manufacturer"}, "Size": 1, "Length": 10.0, "Width": 5.0, "Height": 3.0, "Crew": 1, "Mass": 50000, "IsSpaceship": true, "IsVehicle": false, "IsGravlev": false}
]

View File

@@ -0,0 +1,193 @@
"""API tests for the Star Citizen inventory backend (fixture dataset)."""
from fastapi.testclient import TestClient
from app.api import app
client = TestClient(app)
def test_healthz():
resp = client.get("/healthz")
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "ok"
assert body["counts"] == {
"ships": 2,
"ship-items": 2,
"fps-items": 1,
"commodities": 4,
}
def test_stats():
resp = client.get("/api/v1/stats")
assert resp.status_code == 200
body = resp.json()
assert body["categories"] == {
"ships": 2,
"ship-items": 2,
"fps-items": 1,
"commodities": 4,
}
assert body["manufacturers"] == 4
def test_manufacturers_sorted_by_name():
resp = client.get("/api/v1/manufacturers")
assert resp.status_code == 200
names = [m["name"] for m in resp.json()]
assert names == ["Anvil Aerospace", "Behring", "PHLD", "Unknown Manufacturer"]
def test_items_default_pagination():
resp = client.get("/api/v1/items")
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 9
assert body["page"] == 1
assert body["page_size"] == 25
assert body["total_pages"] == 1
assert len(body["items"]) == 9
def test_items_page_size_capped():
resp = client.get("/api/v1/items", params={"page_size": 500})
assert resp.status_code == 200
assert resp.json()["page_size"] == 100
def test_items_page_size_minimum_enforced():
resp = client.get("/api/v1/items", params={"page_size": 0})
assert resp.status_code == 422 # pydantic ge=1 validation
def test_items_category_filter():
resp = client.get("/api/v1/items", params={"category": "ships"})
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 2
assert all(i["category"] == "ships" for i in body["items"])
def test_items_invalid_category_rejected():
resp = client.get("/api/v1/items", params={"category": "bogus"})
assert resp.status_code == 422
def test_items_search_matches_name_case_insensitive():
resp = client.get("/api/v1/items", params={"search": "carrack"})
body = resp.json()
assert body["total"] == 1
assert body["items"][0]["name"] == "Anvil Carrack"
def test_items_search_matches_class_name():
resp = client.get("/api/v1/items", params={"search": "test_ph"})
body = resp.json()
assert body["total"] == 1
assert body["items"][0]["class_name"] == "TEST_Ph_Item"
def test_items_manufacturer_filter_case_insensitive():
resp = client.get("/api/v1/items", params={"manufacturer": "behr"})
body = resp.json()
assert body["total"] == 2 # Behring shield + Behring rifle
assert all(i["manufacturer"]["code"] == "BEHR" for i in body["items"])
def test_items_type_filter_case_insensitive():
resp = client.get("/api/v1/items", params={"type": "weaponpersonal"})
body = resp.json()
assert body["total"] == 1
assert body["items"][0]["class_name"] == "BEHR_Rifle_Ballistic_01"
def test_items_size_filter():
resp = client.get("/api/v1/items", params={"size": 2})
body = resp.json()
assert body["total"] == 2 # TEST_Ph_Item (size 2) + rifle (size 2)
assert {i["class_name"] for i in body["items"]} == {
"TEST_Ph_Item",
"BEHR_Rifle_Ballistic_01",
}
def test_item_detail_ship():
resp = client.get("/api/v1/items/ANVL_Carrack")
assert resp.status_code == 200
body = resp.json()
assert body["name"] == "Anvil Carrack"
assert body["category"] == "ships"
assert body["details"]["crew"] == 4
assert body["details"]["is_spaceship"] is True
assert body["details"]["length"] == 35.0
def test_item_detail_item_ports():
resp = client.get("/api/v1/items/BEHR_Rifle_Ballistic_01")
assert resp.status_code == 200
body = resp.json()
assert body["details"]["ports"][0]["PortName"] == "magazine_attach"
assert body["details"]["mass"] == 4
def test_item_detail_404():
resp = client.get("/api/v1/items/NOPE_Does_Not_Exist")
assert resp.status_code == 404
assert resp.json()["detail"] == "Item 'NOPE_Does_Not_Exist' not found"
def test_placeholder_name_falls_back_to_class_name():
# Ship with Name == "<= PLACEHOLDER =>"
resp = client.get("/api/v1/items/TEST_Placeholder_Ship")
assert resp.status_code == 200
assert resp.json()["name"] == "TEST Placeholder Ship"
# Item with stdItem.Name == "<= PLACEHOLDER =>"
resp = client.get("/api/v1/items/TEST_Ph_Item")
assert resp.status_code == 200
assert resp.json()["name"] == "TEST Ph Item"
def test_manufacturer_placeholder_name_falls_back_to_code():
# manufacturers.json entry with Name == "<= PLACEHOLDER =>"
resp = client.get("/api/v1/manufacturers")
phld = [m for m in resp.json() if m["code"] == "PHLD"]
assert len(phld) == 1
assert phld[0]["name"] == "PHLD"
# Item whose embedded manufacturer Name is a placeholder
resp = client.get("/api/v1/items/TEST_Ph_Item")
assert resp.json()["manufacturer"]["name"] == "UNKN"
def test_deterministic_sort_order():
body = client.get("/api/v1/items", params={"page_size": 100}).json()
keys = [(i["category"], i["name"].lower()) for i in body["items"]]
assert keys == sorted(keys)
def test_items_category_filter_commodities():
resp = client.get("/api/v1/items", params={"category": "commodities"})
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 4
assert all(i["category"] == "commodities" for i in body["items"])
def test_items_search_matches_refined_and_ore_iron():
resp = client.get("/api/v1/items", params={"search": "iron"})
body = resp.json()
assert body["total"] == 2
assert {i["name"] for i in body["items"]} == {"Iron", "Iron (Ore)"}
def test_item_detail_commodity():
resp = client.get("/api/v1/items/Carryable_TBO_FL_1SCU_Commodity_Metal_Ore_Iron")
assert resp.status_code == 200
body = resp.json()
assert body["name"] == "Iron (Ore)"
assert body["category"] == "commodities"
assert body["type"] == "Cargo"
assert body["size"] is None
assert isinstance(body["details"], dict)

View File

@@ -0,0 +1,151 @@
"""Authentication tests: registration, login, sessions, throttling, CSRF."""
from fastapi.testclient import TestClient
from app.api import app
PASSWORD = "correct horse battery staple"
def new_client() -> TestClient:
return TestClient(app)
def register(client: TestClient, username: str = "alice", password: str = PASSWORD):
return client.post(
"/api/v1/auth/register", json={"username": username, "password": password}
)
def login(client: TestClient, username: str = "alice", password: str = PASSWORD):
return client.post(
"/api/v1/auth/login", json={"username": username, "password": password}
)
def test_register_returns_user_and_authenticates():
client = new_client()
resp = register(client)
assert resp.status_code == 201
body = resp.json()
assert body["user"]["username"] == "alice"
assert "password" not in body["user"]
assert client.cookies.get("sc_session")
me = client.get("/api/v1/auth/me")
assert me.status_code == 200
assert me.json()["user"]["username"] == "alice"
def test_register_duplicate_username_is_conflict_case_insensitive():
client = new_client()
assert register(client, "Alice").status_code == 201
assert register(new_client(), "alice").status_code == 409
def test_register_rejects_weak_password():
resp = register(new_client(), password="short")
assert resp.status_code == 422
def test_register_rejects_invalid_username():
for username in ["ab", "has space", "bad!", "x" * 33]:
assert register(new_client(), username).status_code == 422
def test_me_requires_authentication():
resp = new_client().get("/api/v1/auth/me")
assert resp.status_code == 401
def test_login_is_case_insensitive_and_works_after_logout():
client = new_client()
assert register(client, "alice").status_code == 201
assert client.post("/api/v1/auth/logout").status_code == 204
assert client.get("/api/v1/auth/me").status_code == 401
resp = login(client, "ALICE")
assert resp.status_code == 200
assert resp.json()["user"]["username"] == "alice"
assert client.get("/api/v1/auth/me").status_code == 200
def test_wrong_password_and_unknown_user_are_indistinguishable():
client = new_client()
assert register(client, "alice").status_code == 201
wrong = login(new_client(), "alice", "definitely wrong")
unknown = login(new_client(), "nobody", "definitely wrong")
assert wrong.status_code == unknown.status_code == 401
assert wrong.json()["detail"] == unknown.json()["detail"]
def test_logout_invalidates_session_token():
client = new_client()
assert register(client).status_code == 201
token = client.cookies.get("sc_session")
assert client.post("/api/v1/auth/logout").status_code == 204
# Replay the old cookie on a fresh client: it must be rejected.
replay = new_client()
replay.cookies.set("sc_session", token)
assert replay.get("/api/v1/auth/me").status_code == 401
def test_login_is_rate_limited():
client = new_client()
assert register(client, "alice").status_code == 201
for _ in range(5):
assert login(new_client(), "alice", "wrong password").status_code == 401
blocked = login(new_client(), "alice", PASSWORD)
assert blocked.status_code == 429
assert int(blocked.headers["Retry-After"]) > 0
def test_cross_origin_mutation_is_rejected():
client = new_client()
resp = client.post(
"/api/v1/auth/register",
json={"username": "alice", "password": PASSWORD},
headers={"Origin": "http://evil.example"},
)
assert resp.status_code == 403
def test_same_origin_mutation_is_allowed():
client = new_client()
resp = client.post(
"/api/v1/auth/register",
json={"username": "alice", "password": PASSWORD},
headers={"Origin": "http://testserver"},
)
assert resp.status_code == 201
def test_same_origin_with_forwarded_port_is_allowed():
client = new_client()
resp = client.post(
"/api/v1/auth/register",
json={"username": "alice", "password": PASSWORD},
headers={
"Origin": "http://testserver:8080",
"X-Forwarded-Host": "testserver:8080",
},
)
assert resp.status_code == 201
def test_forwarded_host_without_port_is_rejected():
client = new_client()
resp = client.post(
"/api/v1/auth/register",
json={"username": "alice", "password": PASSWORD},
headers={
"Origin": "http://testserver:8080",
"X-Forwarded-Host": "testserver",
},
)
assert resp.status_code == 403

View File

@@ -0,0 +1,156 @@
"""Public available-items listing: open access, owner attribution, no notes."""
from fastapi.testclient import TestClient
from app.api import app
PASSWORD = "correct horse battery staple"
CARROCK = "ANVL_Carrack"
RIFLE = "BEHR_Rifle_Ballistic_01"
IRON_ORE = "Carryable_TBO_FL_1SCU_Commodity_Metal_Ore_Iron"
CONSTRUCTION = "Carryable_TBO_FL_1SCU_Commodity_ProcessedGoods_ConstructionMaterials"
def authed_client(username: str) -> TestClient:
client = TestClient(app)
resp = client.post(
"/api/v1/auth/register", json={"username": username, "password": PASSWORD}
)
assert resp.status_code == 201, resp.text
return client
def add(
client: TestClient,
class_name: str,
quantity: int = 1,
notes: str = "",
quality: int | None = None,
):
payload = {"class_name": class_name, "quantity": quantity, "notes": notes}
if quality is not None:
payload["quality"] = quality
resp = client.post("/api/v1/me/inventory", json=payload)
assert resp.status_code in (200, 201), resp.text
return resp
def offer(client: TestClient, class_name: str, quantity: int = 1):
add(client, class_name, quantity)
resp = client.patch(
f"/api/v1/me/inventory/{class_name}", json={"visibility": "available"}
)
assert resp.status_code == 200, resp.text
def public_items(**params):
return TestClient(app).get("/api/v1/available", params=params).json()
def test_listing_is_public_and_empty_by_default():
body = public_items()
assert body["total"] == 0
assert body["items"] == []
assert body["page"] == 1
def test_private_items_are_not_listed_until_marked_available():
alice = authed_client("alice")
add(alice, CARROCK)
assert public_items()["total"] == 0
alice.patch(f"/api/v1/me/inventory/{CARROCK}", json={"visibility": "available"})
body = public_items()
assert body["total"] == 1
assert body["items"][0]["class_name"] == CARROCK
alice.patch(f"/api/v1/me/inventory/{CARROCK}", json={"visibility": "private"})
assert public_items()["total"] == 0
def test_listing_shows_owner_and_catalog_details_but_not_notes():
alice = authed_client("alice")
offer(alice, CARROCK, quantity=3)
alice.patch(f"/api/v1/me/inventory/{CARROCK}", json={"notes": "hidden stash"})
item = public_items()["items"][0]
assert item["username"] == "alice"
assert item["name"] == "Anvil Carrack"
assert item["category"] == "ships"
assert item["manufacturer"]["code"] == "ANVL"
assert item["quantity"] == 3
assert "notes" not in item
assert "hidden stash" not in str(public_items())
def test_each_owner_appears_as_its_own_offer():
offer(authed_client("alice"), CARROCK, quantity=1)
offer(authed_client("bob"), CARROCK, quantity=2)
items = public_items(sort="quantity")["items"]
assert [(i["username"], i["quantity"], i["class_name"]) for i in items] == [
("bob", 2, CARROCK),
("alice", 1, CARROCK),
]
def test_removing_an_offer_removes_it_from_the_listing():
alice = authed_client("alice")
offer(alice, CARROCK)
assert public_items()["total"] == 1
assert alice.delete(f"/api/v1/me/inventory/{CARROCK}").status_code == 204
assert public_items()["total"] == 0
assert public_items()["items"] == []
def test_listing_filters_by_category_and_searches_owner_or_item():
offer(authed_client("alice"), CARROCK)
offer(authed_client("bob"), RIFLE)
assert [i["class_name"] for i in public_items(category="ships")["items"]] == [CARROCK]
assert [i["username"] for i in public_items(search="bob")["items"]] == ["bob"]
assert [i["class_name"] for i in public_items(search="carrack")["items"]] == [CARROCK]
def test_listing_paginates_and_rejects_unknown_sort():
for quantity, class_name in ((1, CARROCK), (2, RIFLE)):
offer(authed_client(f"pilot{quantity}"), class_name, quantity=quantity)
body = public_items(page=2, page_size=1)
assert body["total"] == 2
assert body["total_pages"] == 2
assert len(body["items"]) == 1
assert public_items(page_size=500)["page_size"] == 100
assert TestClient(app).get("/api/v1/available?sort=bogus").status_code == 422
def test_commodity_offers_are_listed_with_owner():
alice = authed_client("alice")
offer(alice, IRON_ORE, quantity=12)
body = public_items()
assert body["total"] == 1
item = body["items"][0]
assert item["name"] == "Iron (Ore)"
assert item["category"] == "commodities"
assert item["quantity"] == 12
assert item["username"] == "alice"
def test_commodity_quality_is_listed_publicly():
alice = authed_client("alice")
offer(alice, IRON_ORE, quantity=12)
add(alice, IRON_ORE, quality=875)
item = public_items()["items"][0]
assert item["quality"] == 875
assert item["quality_applicable"] is True
def test_non_quality_commodity_is_listed_without_quality():
alice = authed_client("alice")
offer(alice, CONSTRUCTION)
item = public_items()["items"][0]
assert item["quality"] is None
assert item["quality_applicable"] is False

View File

@@ -0,0 +1,356 @@
"""Personal inventory tests: auth gating, CRUD, isolation, catalog immutability."""
from fastapi.testclient import TestClient
from app.api import app
PASSWORD = "correct horse battery staple"
CARROCK = "ANVL_Carrack"
RIFLE = "BEHR_Rifle_Ballistic_01"
IRON_ORE = "Carryable_TBO_FL_1SCU_Commodity_Metal_Ore_Iron"
IRON = "Carryable_TBO_FL_1SCU_Commodity_Metal_Iron"
CONSTRUCTION = "Carryable_TBO_FL_1SCU_Commodity_ProcessedGoods_ConstructionMaterials"
def authed_client(username: str = "alice") -> TestClient:
client = TestClient(app)
resp = client.post(
"/api/v1/auth/register", json={"username": username, "password": PASSWORD}
)
assert resp.status_code == 201, resp.text
return client
def add(
client: TestClient,
class_name: str,
quantity: int = 1,
notes: str = "",
quality: int | None = None,
):
payload = {"class_name": class_name, "quantity": quantity, "notes": notes}
if quality is not None:
payload["quality"] = quality
return client.post("/api/v1/me/inventory", json=payload)
def test_new_account_starts_with_empty_inventory():
client = authed_client("freshpilot")
listing = client.get("/api/v1/me/inventory").json()
assert listing["total"] == 0
assert listing["items"] == []
assert client.get("/api/v1/me/inventory/summary").json() == {
"items": 0,
"quantity": 0,
"categories": {},
"unavailable": 0,
}
def test_inventory_requires_authentication():
client = TestClient(app)
assert client.get("/api/v1/me/inventory").status_code == 401
assert add(client, CARROCK).status_code == 401
assert client.get("/api/v1/me/inventory/summary").status_code == 401
def test_add_unknown_catalog_item_is_not_found():
client = authed_client()
assert add(client, "NOPE_Does_Not_Exist").status_code == 404
def test_add_creates_enriched_entry():
client = authed_client()
resp = add(client, CARROCK)
assert resp.status_code == 201
body = resp.json()
assert body["created"] is True
entry = body["entry"]
assert entry["class_name"] == CARROCK
assert entry["name"] == "Anvil Carrack"
assert entry["category"] == "ships"
assert entry["quantity"] == 1
assert entry["visibility"] == "private"
assert entry["available"] is True
assert entry["manufacturer"]["code"] == "ANVL"
def test_add_same_item_increments_quantity():
client = authed_client()
assert add(client, CARROCK, quantity=1).status_code == 201
resp = add(client, CARROCK, quantity=2, notes="pledged")
assert resp.status_code == 200
body = resp.json()
assert body["created"] is False
assert body["entry"]["quantity"] == 3
assert body["entry"]["notes"] == "pledged"
listing = client.get("/api/v1/me/inventory").json()
assert listing["total"] == 1
def test_inventories_are_isolated_between_users():
alice = authed_client("alice")
bob = authed_client("bob")
assert add(alice, CARROCK).status_code == 201
assert add(bob, RIFLE).status_code == 201
alice_items = alice.get("/api/v1/me/inventory").json()["items"]
bob_items = bob.get("/api/v1/me/inventory").json()["items"]
assert [i["class_name"] for i in alice_items] == [CARROCK]
assert [i["class_name"] for i in bob_items] == [RIFLE]
def test_patch_updates_quantity_and_notes():
client = authed_client()
assert add(client, CARROCK).status_code == 201
resp = client.patch(
f"/api/v1/me/inventory/{CARROCK}", json={"quantity": 4, "notes": "hangar 3"}
)
assert resp.status_code == 200
body = resp.json()
assert body["quantity"] == 4
assert body["notes"] == "hangar 3"
def test_visibility_toggles_between_private_and_available():
client = authed_client()
assert add(client, CARROCK).status_code == 201
resp = client.patch(
f"/api/v1/me/inventory/{CARROCK}", json={"visibility": "available"}
)
assert resp.status_code == 200
assert resp.json()["visibility"] == "available"
listed = client.get("/api/v1/me/inventory").json()["items"]
assert listed[0]["visibility"] == "available"
back = client.patch(
f"/api/v1/me/inventory/{CARROCK}", json={"visibility": "private"}
)
assert back.json()["visibility"] == "private"
def test_patch_rejects_unknown_visibility():
client = authed_client()
assert add(client, CARROCK).status_code == 201
resp = client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"visibility": "public"})
assert resp.status_code == 422
def test_visibility_is_kept_when_quantity_changes_and_item_is_readded():
client = authed_client()
add(client, CARROCK)
client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"visibility": "available"})
client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"quantity": 3})
assert add(client, CARROCK, quantity=1).status_code == 200
entry = client.get("/api/v1/me/inventory").json()["items"][0]
assert entry["quantity"] == 4
assert entry["visibility"] == "available"
def test_patch_unknown_entry_is_not_found():
client = authed_client()
assert client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"quantity": 2}).status_code == 404
def test_patch_rejects_zero_quantity():
client = authed_client()
assert add(client, CARROCK).status_code == 201
resp = client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"quantity": 0})
assert resp.status_code == 422
def test_delete_removes_entry():
client = authed_client()
assert add(client, CARROCK).status_code == 201
assert client.delete(f"/api/v1/me/inventory/{CARROCK}").status_code == 204
assert client.get("/api/v1/me/inventory").json()["total"] == 0
assert client.delete(f"/api/v1/me/inventory/{CARROCK}").status_code == 404
def test_list_filters_by_category_and_search():
client = authed_client()
assert add(client, CARROCK).status_code == 201
assert add(client, RIFLE).status_code == 201
ships = client.get("/api/v1/me/inventory", params={"category": "ships"}).json()
assert ships["total"] == 1
assert ships["items"][0]["class_name"] == CARROCK
found = client.get("/api/v1/me/inventory", params={"search": "carrack"}).json()
assert found["total"] == 1
assert found["items"][0]["class_name"] == CARROCK
def test_list_pagination_envelope_and_cap():
client = authed_client()
assert add(client, CARROCK).status_code == 201
assert add(client, RIFLE).status_code == 201
body = client.get(
"/api/v1/me/inventory", params={"page_size": 1, "page": 2}
).json()
assert body["total"] == 2
assert body["page"] == 2
assert body["page_size"] == 1
assert body["total_pages"] == 2
assert len(body["items"]) == 1
capped = client.get("/api/v1/me/inventory", params={"page_size": 500}).json()
assert capped["page_size"] == 100
def test_summary_aggregates_quantity_and_categories():
client = authed_client()
assert add(client, CARROCK, quantity=2).status_code == 201
assert add(client, RIFLE, quantity=3).status_code == 201
summary = client.get("/api/v1/me/inventory/summary").json()
assert summary["items"] == 2
assert summary["quantity"] == 5
assert summary["unavailable"] == 0
assert summary["categories"]["ships"] == 2
def test_notes_length_is_bounded():
client = authed_client()
resp = add(client, CARROCK, notes="x" * 1001)
assert resp.status_code == 422
def test_catalog_is_not_mutated_by_inventory_operations():
client = authed_client()
before = client.get("/api/v1/items", params={"page_size": 100}).json()
assert add(client, CARROCK).status_code == 201
client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"quantity": 9})
client.delete(f"/api/v1/me/inventory/{CARROCK}")
after = client.get("/api/v1/items", params={"page_size": 100}).json()
assert after == before
carrack = client.get(f"/api/v1/items/{CARROCK}").json()
assert "quantity" not in carrack
assert "notes" not in carrack
def test_sort_by_category_groups_categories():
client = authed_client()
add(client, CARROCK)
add(client, RIFLE)
items = client.get("/api/v1/me/inventory?sort=category").json()["items"]
assert [i["class_name"] for i in items] == [RIFLE, CARROCK]
def test_sort_by_quantity_is_descending():
client = authed_client()
add(client, CARROCK, quantity=5)
add(client, RIFLE, quantity=1)
items = client.get("/api/v1/me/inventory?sort=quantity").json()["items"]
assert [i["class_name"] for i in items] == [CARROCK, RIFLE]
def test_sort_defaults_to_name_and_rejects_unknown_values():
client = authed_client()
add(client, CARROCK)
listing = client.get("/api/v1/me/inventory").json()
assert [i["class_name"] for i in listing["items"]] == [CARROCK]
assert client.get("/api/v1/me/inventory?sort=bogus").status_code == 422
def test_add_commodity_uses_catalog_name_and_quantity():
client = authed_client("refiner")
resp = add(client, IRON_ORE, quantity=12, notes="from Aaron Halo")
assert resp.status_code == 201
entry = client.get("/api/v1/me/inventory").json()["items"][0]
assert entry["name"] == "Iron (Ore)"
assert entry["category"] == "commodities"
assert entry["quantity"] == 12
assert entry["available"] is True
assert entry["visibility"] == "private"
def test_catalog_marks_only_mineable_commodities_as_quality_applicable():
client = TestClient(app)
items = client.get(
"/api/v1/items", params={"category": "commodities", "page_size": 100}
).json()["items"]
flags = {item["class_name"]: item["quality_applicable"] for item in items}
assert flags[IRON] is True
assert flags[IRON_ORE] is True
assert flags[CONSTRUCTION] is False
assert flags["Carryable_TBO_FL_1SCU_Commodity_Metal_RMC"] is False
ships = client.get("/api/v1/items", params={"category": "ships"}).json()["items"]
assert all(item["quality_applicable"] is False for item in ships)
def test_quality_can_be_set_and_cleared_on_a_quality_commodity():
client = authed_client()
assert add(client, IRON_ORE).status_code == 201
resp = client.patch(f"/api/v1/me/inventory/{IRON_ORE}", json={"quality": 750})
assert resp.status_code == 200
body = resp.json()
assert body["quality"] == 750
assert body["quality_applicable"] is True
cleared = client.patch(
f"/api/v1/me/inventory/{IRON_ORE}", json={"quality": None}
)
assert cleared.status_code == 200
assert cleared.json()["quality"] is None
def test_add_persists_quality_and_keeps_it_when_quantity_is_incremented():
client = authed_client()
first = add(client, IRON, quality=900)
assert first.status_code == 201
assert first.json()["entry"]["quality"] == 900
second = add(client, IRON, quantity=2)
assert second.status_code == 200
entry = second.json()["entry"]
assert entry["quality"] == 900
assert entry["quantity"] == 3
replaced = add(client, IRON, quality=400)
assert replaced.json()["entry"]["quality"] == 400
def test_quality_is_rejected_for_non_mineable_commodities():
client = authed_client()
assert add(client, CONSTRUCTION).status_code == 201
resp = client.patch(
f"/api/v1/me/inventory/{CONSTRUCTION}", json={"quality": 500}
)
assert resp.status_code == 422
assert "does not apply" in resp.json()["detail"]
add_resp = add(client, CONSTRUCTION, quality=500)
assert add_resp.status_code == 422
def test_quality_is_rejected_for_ships():
client = authed_client()
assert add(client, CARROCK).status_code == 201
resp = client.patch(f"/api/v1/me/inventory/{CARROCK}", json={"quality": 100})
assert resp.status_code == 422
def test_quality_range_is_enforced():
client = authed_client()
assert add(client, IRON_ORE, quality=1).status_code == 201
assert add(client, IRON, quality=1000).status_code == 201
assert add(client, IRON_ORE, quality=0).status_code == 422
assert add(client, IRON_ORE, quality=1001).status_code == 422
assert add(client, IRON, quantity=1).status_code == 200
assert client.patch(
f"/api/v1/me/inventory/{IRON}", json={"quality": 1001}
).status_code == 422

View File

@@ -0,0 +1,176 @@
"""Public order board: authenticated creation, public read, owner-only edits."""
from fastapi.testclient import TestClient
from app.api import app
PASSWORD = "correct horse battery staple"
CARROCK = "ANVL_Carrack"
RIFLE = "BEHR_Rifle_Ballistic_01"
IRON_ORE = "Carryable_TBO_FL_1SCU_Commodity_Metal_Ore_Iron"
CONSTRUCTION = "Carryable_TBO_FL_1SCU_Commodity_ProcessedGoods_ConstructionMaterials"
def authed_client(username: str) -> TestClient:
client = TestClient(app)
resp = client.post(
"/api/v1/auth/register", json={"username": username, "password": PASSWORD}
)
assert resp.status_code == 201, resp.text
return client
def place(
client: TestClient,
class_name: str,
quantity: int = 1,
notes: str = "",
quality: int | None = None,
):
payload = {"class_name": class_name, "quantity": quantity, "notes": notes}
if quality is not None:
payload["quality"] = quality
return client.post("/api/v1/orders", json=payload)
def public_orders(**params):
return TestClient(app).get("/api/v1/orders", params=params).json()
def test_order_board_is_public_and_empty_by_default():
body = public_orders()
assert body["total"] == 0
assert body["items"] == []
assert body["page"] == 1
def test_placing_an_order_requires_authentication():
assert place(TestClient(app), CARROCK).status_code == 401
def test_placing_an_order_returns_enriched_public_entry():
alice = authed_client("alice")
resp = place(alice, CARROCK, quantity=2, notes="Need a hauler")
assert resp.status_code == 201, resp.text
order = resp.json()
assert order["class_name"] == CARROCK
assert order["name"] == "Anvil Carrack"
assert order["category"] == "ships"
assert order["manufacturer"]["code"] == "ANVL"
assert order["quantity"] == 2
assert order["notes"] == "Need a hauler"
assert order["username"] == "alice"
assert order["status"] == "open"
assert order["available"] is True
public = public_orders()
assert public["total"] == 1
assert public["items"][0]["id"] == order["id"]
assert public["items"][0]["username"] == "alice"
def test_orders_from_every_user_are_visible():
place(authed_client("alice"), CARROCK)
place(authed_client("bob"), RIFLE)
items = public_orders(sort="name")["items"]
assert [(i["username"], i["class_name"]) for i in items] == [
("alice", CARROCK),
("bob", RIFLE),
]
def test_order_quality_is_stored_and_public_for_ore():
alice = authed_client("alice")
resp = place(alice, IRON_ORE, quantity=10, quality=875)
assert resp.status_code == 201, resp.text
assert resp.json()["quality"] == 875
assert resp.json()["quality_applicable"] is True
item = public_orders()["items"][0]
assert item["quality"] == 875
assert item["quality_applicable"] is True
def test_quality_is_rejected_for_non_mineable_items():
alice = authed_client("alice")
assert place(alice, CONSTRUCTION, quality=500).status_code == 422
assert place(alice, CARROCK, quality=100).status_code == 422
assert place(alice, IRON_ORE, quality=1001).status_code == 422
def test_unknown_catalog_item_is_rejected():
assert place(authed_client("alice"), "NOPE_Does_Not_Exist").status_code == 404
def test_listing_filters_by_category_search_and_status():
alice = authed_client("alice")
bob = authed_client("bob")
first = place(alice, CARROCK).json()
place(bob, RIFLE)
place(alice, IRON_ORE, quality=400)
assert [i["class_name"] for i in public_orders(category="ships")["items"]] == [
CARROCK
]
assert [i["username"] for i in public_orders(search="bob")["items"]] == ["bob"]
assert [
i["class_name"] for i in public_orders(search="iron")["items"]
] == [IRON_ORE]
alice.patch(f"/api/v1/orders/{first['id']}", json={"status": "fulfilled"})
assert [i["class_name"] for i in public_orders(status="open")["items"]] == [
IRON_ORE,
RIFLE,
]
assert [i["class_name"] for i in public_orders(status="fulfilled")["items"]] == [
CARROCK
]
def test_listing_paginates_and_rejects_unknown_sort():
alice = authed_client("alice")
place(alice, CARROCK)
place(alice, RIFLE)
body = public_orders(page=2, page_size=1)
assert body["total"] == 2
assert body["total_pages"] == 2
assert len(body["items"]) == 1
assert public_orders(page_size=500)["page_size"] == 100
assert TestClient(app).get("/api/v1/orders?sort=bogus").status_code == 422
assert TestClient(app).get("/api/v1/orders?status=bogus").status_code == 422
def test_only_the_owner_can_edit_or_remove_an_order():
alice = authed_client("alice")
bob = authed_client("bob")
order = place(alice, IRON_ORE, quality=500).json()
assert (
bob.patch(
f"/api/v1/orders/{order['id']}", json={"quantity": 99}
).status_code
== 404
)
assert bob.delete(f"/api/v1/orders/{order['id']}").status_code == 404
updated = alice.patch(
f"/api/v1/orders/{order['id']}",
json={"quantity": 3, "notes": "Updated", "status": "fulfilled"},
)
assert updated.status_code == 200, updated.text
body = updated.json()
assert body["quantity"] == 3
assert body["notes"] == "Updated"
assert body["status"] == "fulfilled"
assert body["quality"] == 500
cleared = alice.patch(
f"/api/v1/orders/{order['id']}", json={"quality": None}
)
assert cleared.status_code == 200, cleared.text
assert cleared.json()["quality"] is None
assert alice.delete(f"/api/v1/orders/{order['id']}").status_code == 204
assert public_orders()["total"] == 0