"""Tests for the Bulk Rate Update feature (v12.14). Validates: - Preview returns expected old/new/delta math (no DB writes). - Apply persists new rates, never touches stock, logs the batch. - Undo restores old rates and flags the batch as undone. - Invalid operations / targets / negative results are rejected or skipped. - History endpoint returns batches in descending order. """ import time import pytest from conftest import BASE_URL # type: ignore[import-not-found] @pytest.fixture(scope="module") def admin_user_id(admin_session): r = admin_session.get(f"{BASE_URL}/api/me/permissions", timeout=20) assert r.status_code == 200 return r.json()["user_id"] def _get_item(session, company_id, item_id): """No GET /items/{id} exists — list and find by id.""" docs = session.get(f"{BASE_URL}/api/items", params={"company_id": company_id}, timeout=20).json() return next((d for d in docs if d["id"] == item_id), None) @pytest.fixture() def test_items(admin_session, default_company_id): """Create 3 throwaway items with known rates, yield their IDs, then clean up.""" ts = int(time.time() * 1000) created = [] for i, sale in enumerate([100, 200, 300]): r = admin_session.post( f"{BASE_URL}/api/items", params={"company_id": default_company_id}, json={ "name": f"BulkRateTest_{ts}_{i}", "code": f"BRT{ts}{i}", "base_unit": "PCS", "unit": "PCS", "category": "BulkRateTestCat", "gst_rate": 18, "sale_price": sale, "purchase_price": sale * 0.7, "mrp": sale * 1.2, "current_stock": 10, "opening_stock": 10, }, timeout=20, ) assert r.status_code == 200, r.text created.append(r.json()) yield created for it in created: admin_session.delete(f"{BASE_URL}/api/items/{it['id']}", timeout=20) @pytest.mark.critical class TestBulkRateUpdate: def test_preview_pct_increase(self, admin_session, default_company_id, test_items): ids = [it["id"] for it in test_items] r = admin_session.post( f"{BASE_URL}/api/items/bulk-rate-update/preview", json={ "company_id": default_company_id, "target_field": "sale_price", "operation": "inc_pct", "value": 10, # +10% "item_ids": ids, }, timeout=20, ) assert r.status_code == 200, r.text body = r.json() assert body["row_count"] == 3 # Old sale prices were 100, 200, 300; +10% → 110, 220, 330 new_by_id = {row["item_id"]: row["new_value"] for row in body["rows"]} assert new_by_id[test_items[0]["id"]] == 110.0 assert new_by_id[test_items[1]["id"]] == 220.0 assert new_by_id[test_items[2]["id"]] == 330.0 # Preview must NOT mutate items chk = _get_item(admin_session, default_company_id, test_items[0]["id"]) assert chk is not None assert chk["sale_price"] == 100.0 def test_apply_set_absolute_logs_batch_and_keeps_stock(self, admin_session, default_company_id, test_items): ids = [it["id"] for it in test_items] r = admin_session.post( f"{BASE_URL}/api/items/bulk-rate-update/apply", json={ "company_id": default_company_id, "target_field": "purchase_price", "operation": "set", "value": 50, "item_ids": ids, "note": "auto-test set 50", }, timeout=30, ) assert r.status_code == 200, r.text body = r.json() assert body["applied"] == 3 batch_id = body["batch_id"] # Item rates updated; stock unchanged for it in test_items: chk = _get_item(admin_session, default_company_id, it["id"]) assert chk is not None assert chk["purchase_price"] == 50.0 assert chk["current_stock"] == 10 # stock untouched # History contains this batch as top entry hist = admin_session.get( f"{BASE_URL}/api/items/bulk-rate-update/history", params={"company_id": default_company_id, "limit": 5}, timeout=20, ) assert hist.status_code == 200 rows = hist.json() assert any(row["batch_id"] == batch_id and row["row_count"] == 3 for row in rows) def test_apply_then_undo_restores_old_values(self, admin_session, default_company_id, test_items): ids = [it["id"] for it in test_items] # +20% on MRP r = admin_session.post( f"{BASE_URL}/api/items/bulk-rate-update/apply", json={ "company_id": default_company_id, "target_field": "mrp", "operation": "inc_pct", "value": 20, "item_ids": ids, }, timeout=30, ) assert r.status_code == 200, r.text batch_id = r.json()["batch_id"] # Verify MRP changed (was 120/240/360 → 144/288/432) chk0 = _get_item(admin_session, default_company_id, test_items[0]["id"]) assert chk0 is not None assert chk0["mrp"] == 144.0 # Undo u = admin_session.post(f"{BASE_URL}/api/items/bulk-rate-update/undo/{batch_id}", timeout=30) assert u.status_code == 200, u.text assert u.json()["reverted"] == 3 chk_after = _get_item(admin_session, default_company_id, test_items[0]["id"]) assert chk_after["mrp"] == 120.0 # restored # Re-undo must 409 u2 = admin_session.post(f"{BASE_URL}/api/items/bulk-rate-update/undo/{batch_id}", timeout=30) assert u2.status_code == 409 def test_skip_when_result_would_be_zero_or_negative(self, admin_session, default_company_id, test_items): # 100% decrease → 0; must be skipped r = admin_session.post( f"{BASE_URL}/api/items/bulk-rate-update/preview", json={ "company_id": default_company_id, "target_field": "sale_price", "operation": "dec_pct", "value": 100, "item_ids": [test_items[0]["id"]], }, timeout=20, ) assert r.status_code == 200, r.text body = r.json() assert body["row_count"] == 0 assert body["skipped_count"] == 1 assert "must be > 0" in body["skipped"][0]["reason"] def test_invalid_operation_rejected(self, admin_session, default_company_id, test_items): r = admin_session.post( f"{BASE_URL}/api/items/bulk-rate-update/preview", json={ "company_id": default_company_id, "target_field": "sale_price", "operation": "multiply", # not supported "value": 2, "item_ids": [test_items[0]["id"]], }, timeout=20, ) assert r.status_code == 400, r.text assert "operation" in r.text.lower() def test_category_filter_applies_to_all_matching(self, admin_session, default_company_id, test_items): """All three test items share category 'BulkRateTestCat' — filter should match all.""" r = admin_session.post( f"{BASE_URL}/api/items/bulk-rate-update/preview", json={ "company_id": default_company_id, "target_field": "sale_price", "operation": "inc_abs", "value": 5, "filters": {"category": "BulkRateTestCat"}, }, timeout=20, ) assert r.status_code == 200, r.text body = r.json() assert body["row_count"] >= 3, f"Expected ≥3 rows for category filter, got {body['row_count']}"