"""Phase 5 — Nursery cost accounting / production costing tests.""" from datetime import date from decimal import Decimal import pytest from app.models.finance import JournalEntry, Account from app.models.accounting import ProductionCostEntry, InventoryLot from app.models.inventory import InventoryItem from app.models.passport import PlantPassport def _codes(db): return {a.id: a.code for a in db.query(Account).all()} def _entry_lines(db, entry): codes = _codes(db) return {(codes[ln.account_id], "debit" if Decimal(str(ln.debit)) > 0 else "credit", Decimal(str(ln.debit)) if Decimal(str(ln.debit)) > 0 else Decimal(str(ln.credit))) for ln in entry.lines} def _make_plant(client, auth_headers, name="Rose", retail_price=100): r = client.post("/api/plants", json={"name": name, "retail_price": retail_price}, headers=auth_headers) assert r.status_code == 201, r.text return r.json() def _make_batch(client, auth_headers, plant_id, qty): r = client.post("/api/production/batches", json={ "plant_id": plant_id, "sowing_date": str(date.today()), "expected_ready_date": str(date.today().replace(year=date.today().year + 1)), "sown_quantity": qty, "current_quantity": qty, }, headers=auth_headers) assert r.status_code == 201, r.text return r.json() class TestBatchCostPosting: def test_cost_from_cash_posts_wip_debit(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 100) r = client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "labor", "amount": 500, "quantity": 100, "unit": "plant", "notes": "Planting crew", }, headers=auth_headers) assert r.status_code == 201, r.text assert r.json()["journal_entry_id"] entry = db.query(JournalEntry).filter( JournalEntry.source == "production", JournalEntry.reference_type == "batch_cost", JournalEntry.reference_id == r.json()["id"], ).first() assert entry is not None lines = _entry_lines(db, entry) # WIP 1050 debit against branch cash (1010) assert ("1050", "debit", Decimal("500.00")) in lines assert ("1010", "credit", Decimal("500.00")) in lines summary = client.get( f"/api/accounting/production-batches/{batch['id']}/costing", headers=auth_headers).json() assert summary["posted_cost"] == Decimal("500.00") assert summary["net_wip"] == Decimal("500.00") assert summary["unit_cost"] == Decimal("5.00") assert summary["by_type"]["labor"] == Decimal("500.00") def test_cost_from_funding_account_credits_funding(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 50) funding = db.query(Account).filter(Account.code == "3020").first() r = client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "fertilizer", "amount": 300, "account_id": funding.id, "unit": "plant", "notes": "GRN fert 4-1", }, headers=auth_headers) assert r.status_code == 201, r.text entry = db.query(JournalEntry).filter( JournalEntry.source == "production", JournalEntry.reference_type == "batch_cost", JournalEntry.reference_id == r.json()["id"], ).first() lines = _entry_lines(db, entry) assert ("1050", "debit", Decimal("300.00")) in lines assert ("3020", "credit", Decimal("300.00")) in lines def test_cost_consumes_fifo_inventory(self, client, auth_headers, db): # Seedlings in inventory at 40 each via GRN-like transaction inv = client.post("/api/inventory", json={ "name": "Seedling tray", "category": "seedling", "unit": "piece", "quantity": 0, "cost_price": 40, "sell_price": 60, }, headers=auth_headers).json() item = db.query(InventoryItem).filter(InventoryItem.id == inv["id"]).first() item.quantity = 10 db.commit() plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 100) r = client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "seed", "amount": 400, "quantity": 10, "unit": "tray", "inventory_item_id": inv["id"], "notes": "Trays into batch", }, headers=auth_headers) assert r.status_code == 201, r.text entry = db.query(JournalEntry).filter( JournalEntry.source == "production", JournalEntry.reference_type == "batch_cost", JournalEntry.reference_id == r.json()["id"], ).first() lines = _entry_lines(db, entry) assert ("1050", "debit", Decimal("400.00")) in lines assert ("1045", "credit", Decimal("400.00")) in lines db.refresh(item) # The whole on-hand value (legacy FIFO plan) moved into WIP assert item.quantity == 0 def test_cost_janitorial_rejects_unknown_type(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 10) r = client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "snacks", "amount": 10, }, headers=auth_headers) assert r.status_code == 400 assert "Unsupported cost type" in r.json()["detail"] class TestBatchReady: def test_ready_creates_lot_refreshes_passport(self, client, auth_headers, db): plant = _make_plant(client, auth_headers, retail_price=100) batch = _make_batch(client, auth_headers, plant["id"], 100) client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "labor", "amount": 500, "quantity": 100, "unit": "plant", }, headers=auth_headers) # Passport generated during production gets the heuristic 45.00 cost pr = client.post("/api/passports", json={"batch_id": batch["id"]}, headers=auth_headers) assert pr.status_code == 201, pr.text passport = db.query(PlantPassport).filter(PlantPassport.batch_id == batch["id"]).first() assert passport.production_cost == Decimal("45.00") r = client.post(f"/api/accounting/production-batches/{batch['id']}/ready", json={}, headers=auth_headers) assert r.status_code == 200, r.text assert r.json()["unit_cost"] == Decimal("5.00") assert r.json()["transfer_qty"] == 100 # Journal: inventory 1040 <- WIP 1050 entry = db.query(JournalEntry).filter( JournalEntry.source == "production", JournalEntry.reference_type == "ready", JournalEntry.reference_id == batch["id"], ).first() assert entry is not None lines = _entry_lines(db, entry) assert ("1040", "debit", Decimal("500.00")) in lines assert ("1050", "credit", Decimal("500.00")) in lines # Plant inventory item auto-created with code == plant.code item = db.query(InventoryItem).filter(InventoryItem.code == plant["code"]).first() assert item is not None assert item.quantity == 100 lot = db.query(InventoryLot).filter(InventoryLot.item_id == item.id, InventoryLot.batch_id == batch["id"]).first() assert lot is not None assert lot.remaining_qty == 100 assert lot.unit_cost == Decimal("5.00") # Passport refresh overrides the heuristic 45.00 with the real unit cost db.refresh(passport) assert passport.production_cost == Decimal("5.00") assert passport.profit_margin == Decimal("95.00") def test_double_ready_blocked(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 20) client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "labor", "amount": 100, "quantity": 20, "unit": "plant", }, headers=auth_headers) client.post(f"/api/accounting/production-batches/{batch['id']}/ready", json={}, headers=auth_headers) r = client.post(f"/api/accounting/production-batches/{batch['id']}/ready", json={}, headers=auth_headers) assert r.status_code == 400 assert "already" in r.json()["detail"].lower() def test_ready_with_zero_wip_rejected(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 10) r = client.post(f"/api/accounting/production-batches/{batch['id']}/ready", json={}, headers=auth_headers) assert r.status_code == 400 class TestMortality: def test_mortality_before_ready_hits_wip(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 100) client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "labor", "amount": 500, "quantity": 100, "unit": "plant", }, headers=auth_headers) r = client.post(f"/api/accounting/production-batches/{batch['id']}/mortality", json={"quantity": 10, "reason": "Cold snap"}, headers=auth_headers) assert r.status_code == 200, r.text assert r.json()["amount"] == Decimal("50.00") entry = db.query(JournalEntry).filter( JournalEntry.source == "production", JournalEntry.reference_type == "mortality", JournalEntry.reference_id == r.json()["entry_id"], ).first() lines = _entry_lines(db, entry) assert ("6130", "debit", Decimal("50.00")) in lines assert ("1050", "credit", Decimal("50.00")) in lines summary = client.get( f"/api/accounting/production-batches/{batch['id']}/costing", headers=auth_headers).json() assert summary["net_wip"] == Decimal("450.00") assert summary["unit_cost"] == Decimal("5.00") assert summary["mortality_qty"] == 10 def test_mortality_after_ready_drains_lot(self, client, auth_headers, db): plant = _make_plant(client, auth_headers) batch = _make_batch(client, auth_headers, plant["id"], 100) client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "labor", "amount": 500, "quantity": 100, "unit": "plant", }, headers=auth_headers) client.post(f"/api/accounting/production-batches/{batch['id']}/ready", json={}, headers=auth_headers) r = client.post(f"/api/accounting/production-batches/{batch['id']}/mortality", json={"quantity": 25, "reason": "Pest"}, headers=auth_headers) assert r.status_code == 200, r.text assert r.json()["amount"] == Decimal("125.00") entry = db.query(JournalEntry).filter( JournalEntry.source == "production", JournalEntry.reference_type == "mortality", JournalEntry.reference_id == r.json()["entry_id"], ).first() lines = _entry_lines(db, entry) assert ("6130", "debit", Decimal("125.00")) in lines assert ("1040", "credit", Decimal("125.00")) in lines detail = client.get(f"/api/production/batches/{batch['id']}", headers=auth_headers).json() assert detail["current_quantity"] == 75 class TestProfitability: def test_ready_batch_sale_then_profitability(self, client, auth_headers, db): plant = _make_plant(client, auth_headers, retail_price=100) batch = _make_batch(client, auth_headers, plant["id"], 100) client.post("/api/accounting/production-costs", json={ "batch_id": batch["id"], "cost_type": "labor", "amount": 500, "quantity": 100, "unit": "plant", }, headers=auth_headers) ready = client.post(f"/api/accounting/production-batches/{batch['id']}/ready", json={}, headers=auth_headers).json() inv_id = ready["inventory_item_id"] # Sale draws down the batch FIFO lot -> COGS at unit cost 5 customer = client.post("/api/customers", json={ "name": "Buyer", "phone": "0182444222", "customer_type": "Retail", }, headers=auth_headers).json() sale = client.post("/api/sales", json={ "customer_id": customer["id"], "paid": 1000.0, "items": [{"plant_id": plant["id"], "inventory_id": inv_id, "quantity": 10, "unit_price": 100}], }, headers=auth_headers) assert sale.status_code == 201, sale.text report = client.get("/api/accounting/batch-profitability", headers=auth_headers).json() assert report["batches"] row = report["batches"][0] assert row["batch_id"] == batch["id"] assert row["unit_cost"] == Decimal("5.00") assert row["sold_qty"] == 10 assert row["revenue"] == Decimal("1000.00") assert row["cost_of_sold"] == Decimal("50.00") assert row["profit"] == Decimal("950.00")