import pytest import httpx import uuid from datetime import date from decimal import Decimal from sqlalchemy.future import select from app.main import app from app.core.database import AsyncSessionLocal from app.models.bom_models import ProductMaster, BomMaster, BomDetail from app.models.rm_models import RmMaster from app.models.yard_models import ChassisMaster from app.modules.yard import service, schemas from tests.conftest import _admin_user # Declare all tests in this file as async pytestmark = pytest.mark.asyncio async def test_bom_flow(): # Setup test entities in DB using direct DB session async with AsyncSessionLocal() as db: # Create temp product test_product = ProductMaster( name="Test Vehicle X1", product_code=f"TEST-PROD-{uuid.uuid4().hex[:6].upper()}", unit_of_measurement="NOS" ) db.add(test_product) # Create temp RMs test_rm1 = RmMaster( name="Test Aluminum Sheet", part_no=f"T-RM-{uuid.uuid4().hex[:6].upper()}", unit_of_measurement="KG" ) test_rm2 = RmMaster( name="Test Steel Plate", part_no=f"T-RM-{uuid.uuid4().hex[:6].upper()}", unit_of_measurement="KG" ) db.add(test_rm1) db.add(test_rm2) await db.commit() await db.refresh(test_product) await db.refresh(test_rm1) await db.refresh(test_rm2) product_id = test_product.product_id rm1_id = test_rm1.rm_id rm2_id = test_rm2.rm_id created_bom_ids = [] try: # Use httpx AsyncClient to test the API endpoints async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as client: # 1. List products and verify our product is present resp = await client.get("/api/v1/bom/products") assert resp.status_code == 200 products_list = resp.json() assert any(p["product_id"] == str(product_id) for p in products_list) # 2. Test create BOM with invalid Product ID (should return 400) invalid_prod_resp = await client.post("/api/v1/bom/", json={ "product_id": str(uuid.uuid4()), "bom_number": f"TEST-BOM-FAIL-{uuid.uuid4().hex[:6].upper()}", "description": "Invalid product", "details": [{"rm_id": str(rm1_id), "quantity": 10.0, "uom": "KG"}] }) assert invalid_prod_resp.status_code == 400 assert "Product not found" in invalid_prod_resp.json()["detail"] # 3. Test create BOM with invalid RM ID (should return 400 Bad Request instead of 500) fake_rm_id = str(uuid.uuid4()) invalid_rm_resp = await client.post("/api/v1/bom/", json={ "product_id": str(product_id), "bom_number": f"TEST-BOM-FAIL-{uuid.uuid4().hex[:6].upper()}", "description": "Invalid RM id test", "details": [ {"rm_id": str(rm1_id), "quantity": 5.0, "uom": "KG"}, {"rm_id": fake_rm_id, "quantity": 12.0, "uom": "KG"} ] }) assert invalid_rm_resp.status_code == 400 assert "Invalid Raw Material ID(s)" in invalid_rm_resp.json()["detail"] assert fake_rm_id in invalid_rm_resp.json()["detail"] # 4. The same raw material twice in one BOM is rejected dup_resp = await client.post("/api/v1/bom/", json={ "product_id": str(product_id), "bom_number": f"T-BOM-{uuid.uuid4().hex[:6].upper()}", "description": "Duplicate RMs test", "details": [ {"rm_id": str(rm1_id), "quantity": 5.0, "uom": "KG"}, {"rm_id": str(rm1_id), "quantity": 10.0, "uom": "KG"} ] }) assert dup_resp.status_code == 400 assert "Duplicate raw materials" in dup_resp.json()["detail"] # 5. Create a valid BOM with distinct components bom_number = f"T-BOM-{uuid.uuid4().hex[:6].upper()}" create_payload = { "product_id": str(product_id), "bom_number": bom_number, "description": "Integration test BOM", "details": [ {"rm_id": str(rm1_id), "quantity": 5.5, "uom": "KG"}, {"rm_id": str(rm2_id), "quantity": 2.2, "uom": "KG"} ] } resp = await client.post("/api/v1/bom/", json=create_payload) assert resp.status_code == 201 bom_data = resp.json() bom_id = bom_data["bom_id"] created_bom_ids.append(bom_id) assert bom_data["bom_number"] == bom_number assert len(bom_data["details"]) == 2 # 6. Retrieve the BOM and check details get_resp = await client.get(f"/api/v1/bom/{bom_id}") assert get_resp.status_code == 200 get_data = get_resp.json() assert get_data["bom_number"] == bom_number assert get_data["product_name"] == "Test Vehicle X1" # 7. Update BOM (valid update, reducing to one component and modifying description) update_payload = { "description": "Updated description", "details": [ {"rm_id": str(rm1_id), "quantity": 7.0, "uom": "KG"} ] } update_resp = await client.put(f"/api/v1/bom/{bom_id}", json=update_payload) assert update_resp.status_code == 200 updated_data = update_resp.json() assert updated_data["description"] == "Updated description" assert len(updated_data["details"]) == 1 assert Decimal(str(updated_data["details"][0]["quantity"])) == Decimal("7.0") # 8. Update BOM with non-existent RM ID (should return 400 Bad Request) update_fail_payload = { "details": [ {"rm_id": fake_rm_id, "quantity": 10.0, "uom": "KG"} ] } update_fail_resp = await client.put(f"/api/v1/bom/{bom_id}", json=update_fail_payload) assert update_fail_resp.status_code == 400 assert "Invalid Raw Material ID(s)" in update_fail_resp.json()["detail"] finally: # Cleanup phase: restore DB to its initial state async with AsyncSessionLocal() as db: # Delete details and BOM masters for created BOM IDs if created_bom_ids: bom_uuids = [uuid.UUID(bid) for bid in created_bom_ids] await db.execute( BomDetail.__table__.delete().where(BomDetail.bom_id.in_(bom_uuids)) ) await db.execute( BomMaster.__table__.delete().where(BomMaster.bom_id.in_(bom_uuids)) ) # Delete test product await db.execute( ProductMaster.__table__.delete().where(ProductMaster.product_id == product_id) ) # Delete test RMs await db.execute( RmMaster.__table__.delete().where(RmMaster.rm_id.in_([rm1_id, rm2_id])) ) await db.commit() async def test_da_lite_remarks_persist_and_expose_in_response(): async with AsyncSessionLocal() as db: chassis = ChassisMaster( chassis_no=f"TEST-DALITE-{uuid.uuid4().hex[:8].upper()}", fuel_type="DIESEL", arrival_date=date.today(), da_lite_status="PENDING", ) db.add(chassis) await db.commit() await db.refresh(chassis) updated = await service.update_da_lite( db, chassis.chassis_id, "HOLD", "Wire pin loose on dongle", (await _admin_user()).user_id, ) assert updated.da_lite_status == "HOLD" assert updated.da_lite_remarks == "Wire pin loose on dongle" serialized = schemas.ChassisOut.model_validate(updated) assert serialized.da_lite_remarks == "Wire pin loose on dongle" await db.delete(chassis) await db.commit()