cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
SZCJW
2026-03-25 23:59:29 +08:00
parent 7964ce1033
commit 23e5df0364
46 changed files with 4168 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+139
View File
@@ -0,0 +1,139 @@
import os
import sys
import tempfile
vendor_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "_vendor"))
if os.path.isdir(vendor_dir) and vendor_dir not in sys.path:
sys.path.insert(0, vendor_dir)
src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))
if os.path.isdir(src_dir) and src_dir not in sys.path:
sys.path.insert(0, src_dir)
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import select
from fastapi import FastAPI, APIRouter
from api.inventory import inventory_router
from models.database import Base, User, Customer, Warehouse, Supplier, Product, ProductMaterial, Inventory
from database.database import get_db_session
from services.auth_service import get_current_active_user
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="session")
def sqlite_db_path():
fd, path = tempfile.mkstemp(prefix="gemold_test_", suffix=".db")
os.close(fd)
yield path
try:
os.remove(path)
except OSError:
pass
@pytest.fixture(scope="session")
async def async_engine(sqlite_db_path):
engine = create_async_engine(f"sqlite+aiosqlite:///{sqlite_db_path}", future=True)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest.fixture(scope="function")
async def db_session(async_engine):
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
async with session_factory() as session:
yield session
await session.rollback()
@pytest.fixture(scope="function")
async def seeded_db(db_session: AsyncSession):
user = User(
id=1,
username="tester",
email="tester@example.com",
hashed_password="x",
full_name="Tester",
is_active=True,
)
customer = Customer(id=1, code="C001", name="客户A", is_active=True)
supplier = Supplier(id=1, code="S001", name="供应商A", is_active=True)
warehouse = Warehouse(id=1, code="W001", name="默认仓库", is_active=True, is_default=True)
material = Product(
id=1,
sku="MAT-001",
name="钢材",
unit="kg",
item_type="material",
cost_price=10.0,
sale_price=0,
min_stock=0,
max_stock=100000,
is_active=True,
)
finished = Product(
id=2,
sku="MOLD-STD",
name="标准模具",
unit="套",
item_type="finished",
cost_price=0,
sale_price=1000.0,
min_stock=0,
max_stock=0,
is_active=True,
)
bom = ProductMaterial(
id=1,
finished_product_id=finished.id,
material_product_id=material.id,
quantity=2.0,
loss_rate=0.05,
)
inv = Inventory(
id=1,
product_id=material.id,
warehouse_id=warehouse.id,
quantity=1000,
locked_quantity=0,
)
db_session.add_all([user, customer, supplier, warehouse, material, finished, bom, inv])
await db_session.commit()
return {"user": user, "customer": customer, "supplier": supplier, "warehouse": warehouse, "material": material, "finished": finished}
@pytest.fixture(scope="function")
async def client(async_engine, seeded_db):
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
test_app = FastAPI()
test_app.include_router(inventory_router)
async def override_get_db_session():
async with session_factory() as session:
yield session
async def override_get_current_active_user():
async with session_factory() as session:
result = await session.execute(select(User).where(User.username == "tester"))
return result.scalar_one()
test_app.dependency_overrides[get_db_session] = override_get_db_session
test_app.dependency_overrides[get_current_active_user] = override_get_current_active_user
transport = ASGITransport(app=test_app, lifespan="off")
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
test_app.dependency_overrides.clear()
tests/test_api_inventory_orders.py
+259
View File
@@ -0,0 +1,259 @@
import pytest
@pytest.mark.anyio
async def test_inventory_list_returns_only_materials(client):
resp = await client.get("/api/inventory")
assert resp.status_code == 200
rows = resp.json()
assert any(r["product_sku"] == "MAT-001" for r in rows)
assert all(r["item_type"] == "material" for r in rows)
@pytest.mark.anyio
async def test_purchase_order_create_and_receive_flow(client):
create_payload = {
"supplier_id": 1,
"expected_date": None,
"remark": "po",
"items": [{"product_id": 1, "quantity": 5, "unit_price": 9999.0, "remark": None}],
}
resp = await client.post("/api/purchase-orders", json=create_payload)
assert resp.status_code == 201
order = resp.json()
assert order["status"] == "draft"
assert order["total_amount"] == 50.0
order_id = order["id"]
detail = await client.get(f"/api/purchase-orders/{order_id}")
assert detail.status_code == 200
item_id = detail.json()["items"][0]["id"]
recv1 = await client.post(
f"/api/purchase-orders/{order_id}/receive",
json={"warehouse_id": 1, "items": [{"item_id": item_id, "receive_quantity": 3}], "remark": "r1"},
)
assert recv1.status_code == 200
assert recv1.json()["status"] == "partial_received"
recv2 = await client.post(
f"/api/purchase-orders/{order_id}/receive",
json={"warehouse_id": 1, "items": [{"item_id": item_id, "receive_quantity": 2}], "remark": "r2"},
)
assert recv2.status_code == 200
assert recv2.json()["status"] == "received"
@pytest.mark.anyio
async def test_purchase_order_receive_overflow_rejected(client):
create_payload = {
"supplier_id": 1,
"items": [{"product_id": 1, "quantity": 2}],
}
resp = await client.post("/api/purchase-orders", json=create_payload)
assert resp.status_code == 201
order_id = resp.json()["id"]
detail = await client.get(f"/api/purchase-orders/{order_id}")
item_id = detail.json()["items"][0]["id"]
recv = await client.post(
f"/api/purchase-orders/{order_id}/receive",
json={"warehouse_id": 1, "items": [{"item_id": item_id, "receive_quantity": 3}]},
)
assert recv.status_code == 400
@pytest.mark.anyio
async def test_purchase_order_update_blocked_after_receive(client):
create_payload = {
"supplier_id": 1,
"items": [{"product_id": 1, "quantity": 2}],
}
resp = await client.post("/api/purchase-orders", json=create_payload)
assert resp.status_code == 201
order_id = resp.json()["id"]
detail = await client.get(f"/api/purchase-orders/{order_id}")
item_id = detail.json()["items"][0]["id"]
recv = await client.post(
f"/api/purchase-orders/{order_id}/receive",
json={"warehouse_id": 1, "items": [{"item_id": item_id, "receive_quantity": 1}]},
)
assert recv.status_code == 200
update_payload = {
"supplier_id": 1,
"items": [{"product_id": 1, "quantity": 2}],
}
upd = await client.put(f"/api/purchase-orders/{order_id}", json=update_payload)
assert upd.status_code == 400
@pytest.mark.anyio
async def test_sales_order_bom_plan_and_auto_issue(client):
create_payload = {
"customer_id": 1,
"items": [{"product_id": 2, "quantity": 1, "unit_price": 1000.0}],
}
resp = await client.post("/api/sales-orders", json=create_payload)
assert resp.status_code == 201
order_id = resp.json()["id"]
plan = await client.get(f"/api/sales-orders/{order_id}/production-plan")
assert plan.status_code == 200
items = plan.json()["items"]
assert len(items) == 1
assert items[0]["material_sku"] == "MAT-001"
assert items[0]["required_quantity"] == 3
assert items[0]["shortage_quantity"] == 0
movements = await client.get("/api/stock-movements", params={"product_id": 1})
assert movements.status_code == 200
assert any(m["movement_type"] == "issue_to_production" for m in movements.json())
@pytest.mark.anyio
async def test_sales_order_create_rejected_when_material_short(client):
inv_list = await client.get("/api/inventory")
inv_id = next(r["inventory_id"] for r in inv_list.json() if r["product_sku"] == "MAT-001")
set_zero = await client.put(
f"/api/inventory/{inv_id}",
json={"quantity": 0, "locked_quantity": 0, "batch_number": None, "location": None},
)
assert set_zero.status_code == 200
create_payload = {
"customer_id": 1,
"items": [{"product_id": 2, "quantity": 1, "unit_price": 1000.0}],
}
resp = await client.post("/api/sales-orders", json=create_payload)
assert resp.status_code == 400
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload",
[
{"customer_id": 1, "items": []},
{"customer_id": 1, "items": [{"product_id": 2, "quantity": 0, "unit_price": 1.0}]},
{"customer_id": 1, "items": [{"product_id": 2, "quantity": 1, "unit_price": -0.01}]},
],
)
async def test_sales_order_schema_validation(payload, client):
resp = await client.post("/api/sales-orders", json=payload)
assert resp.status_code == 422
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload",
[
{"customer_id": 1, "items": [{"product_id": None, "product_sku": None, "product_name": None, "quantity": 1, "unit_price": 1.0}]},
{"customer_id": 1, "items": [{"product_id": 99999, "quantity": 1, "unit_price": 1.0}]},
{"customer_id": 1, "items": [{"product_id": 1, "quantity": 1, "unit_price": 1.0}]},
],
)
async def test_sales_order_item_semantic_validation(payload, client):
resp = await client.post("/api/sales-orders", json=payload)
assert resp.status_code == 400
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload",
[
{"supplier_id": 1, "items": []},
{"supplier_id": 1, "items": [{"product_id": 1, "quantity": 0}]},
{"supplier_id": 99999, "items": [{"product_id": 1, "quantity": 1}]},
{"supplier_id": 1, "items": [{"product_id": 2, "quantity": 1}]},
],
)
async def test_purchase_order_validation(payload, client):
resp = await client.post("/api/purchase-orders", json=payload)
assert resp.status_code in {400, 422}
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload",
[
{"status": "draft"},
{"status": "pending"},
{"status": "received"},
{"status": ""},
],
)
async def test_sales_order_status_enum_rejected(payload, client):
create_payload = {
"customer_id": 1,
"items": [{"product_id": 2, "quantity": 1, "unit_price": 1000.0}],
}
resp = await client.post("/api/sales-orders", json=create_payload)
assert resp.status_code == 201
order_id = resp.json()["id"]
patch = await client.patch(f"/api/sales-orders/{order_id}/status", json=payload)
assert patch.status_code == 400
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload",
[
{"warehouse_id": 1, "items": []},
{"warehouse_id": 1, "items": [{"item_id": 99999, "receive_quantity": 1}]},
{"warehouse_id": 99999, "items": [{"item_id": 1, "receive_quantity": 1}]},
{"warehouse_id": 1, "items": [{"item_id": 1, "receive_quantity": 0}]},
],
)
async def test_purchase_receive_validation(payload, client):
create_payload = {"supplier_id": 1, "items": [{"product_id": 1, "quantity": 1}]}
resp = await client.post("/api/purchase-orders", json=create_payload)
assert resp.status_code == 201
order_id = resp.json()["id"]
detail = await client.get(f"/api/purchase-orders/{order_id}")
assert detail.status_code == 200
real_item_id = detail.json()["items"][0]["id"]
normalized = payload.copy()
if normalized.get("items"):
normalized["items"] = [dict(normalized["items"][0])]
if normalized["items"][0]["item_id"] == 1:
normalized["items"][0]["item_id"] = real_item_id
recv = await client.post(f"/api/purchase-orders/{order_id}/receive", json=normalized)
assert recv.status_code in {400, 404, 422}
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload,expected",
[
({"product_id": 2, "warehouse_id": 1, "quantity": 1, "locked_quantity": 0}, 400),
({"product_id": 1, "warehouse_id": 1, "quantity": -1, "locked_quantity": 0}, 400),
({"product_id": 1, "warehouse_id": 1, "quantity": 1, "locked_quantity": 2}, 400),
({"product_id": 1, "warehouse_id": 99999, "quantity": 1, "locked_quantity": 0}, 404),
],
)
async def test_inventory_create_validation(payload, expected, client):
resp = await client.post("/api/inventory", json=payload)
assert resp.status_code == expected
@pytest.mark.anyio
@pytest.mark.parametrize(
"payload,expected",
[
({"product_id": 1, "warehouse_id": 1, "movement_type": "purchase_in", "quantity": 0}, 400),
({"product_id": 1, "warehouse_id": 1, "movement_type": "issue_to_production", "quantity": 999999}, 400),
({"product_id": 99999, "warehouse_id": 1, "movement_type": "purchase_in", "quantity": 1}, 404),
({"product_id": 1, "warehouse_id": 99999, "movement_type": "purchase_in", "quantity": 1}, 404),
],
)
async def test_stock_movement_create_validation(payload, expected, client):
resp = await client.post("/api/stock-movements", json=payload)
assert resp.status_code == expected
tests/test_sales_order_delivered_freeze.py
+78
View File
@@ -0,0 +1,78 @@
import pytest
@pytest.mark.anyio
async def test_delivered_sales_order_cannot_be_updated_or_deleted(client):
payload = {
"customer_id": 1,
"delivery_date": None,
"remark": "test",
"items": [
{
"product_id": None,
"product_sku": "MOLD-001",
"product_name": "模具001",
"product_category": "模具",
"product_unit": "套",
"quantity": 1,
"unit_price": 1000.0,
"remark": None,
}
],
}
create_resp = await client.post("/api/sales-orders", json=payload)
assert create_resp.status_code == 201
order = create_resp.json()
order_id = order["id"]
status_resp = await client.patch(f"/api/sales-orders/{order_id}/status", json={"status": "delivered"})
assert status_resp.status_code == 200
update_resp = await client.put(f"/api/sales-orders/{order_id}", json=payload)
assert update_resp.status_code == 400
delete_resp = await client.delete(f"/api/sales-orders/{order_id}")
assert delete_resp.status_code == 400
patch_resp = await client.patch(f"/api/sales-orders/{order_id}/status", json={"status": "paid"})
assert patch_resp.status_code == 400
@pytest.mark.anyio
async def test_non_delivered_sales_order_can_be_updated(client):
payload = {
"customer_id": 1,
"delivery_date": None,
"remark": "test",
"items": [
{
"product_id": None,
"product_sku": "MOLD-002",
"product_name": "模具002",
"product_category": "模具",
"product_unit": "套",
"quantity": 1,
"unit_price": 2000.0,
"remark": None,
}
],
}
create_resp = await client.post("/api/sales-orders", json=payload)
assert create_resp.status_code == 201
order_id = create_resp.json()["id"]
payload2 = {
**payload,
"remark": "changed",
"items": [
{
**payload["items"][0],
"quantity": 2,
}
],
}
update_resp = await client.put(f"/api/sales-orders/{order_id}", json=payload2)
assert update_resp.status_code == 200
assert update_resp.json()["total_amount"] == 4000.0