"""v12.3 — Backup Trash (soft-delete + restore + retention) and Migration engine extension (CSV expenses, Tally/Vyapar/Marg aliases, txn-prefixes archive filter).""" import io import os import pytest BASE_URL = os.environ.get("REACT_APP_BACKEND_URL", "https://offline-billing-pro-2.preview.emergentagent.com").rstrip("/") # ============ Backup Trash ============ class TestBackupTrash: @pytest.fixture(scope="class") def backup_id(self, admin_session): r = admin_session.post(f"{BASE_URL}/api/backup/create", json={"label": "qa-trash-test", "encrypt": False}, timeout=60) assert r.status_code == 200, r.text return r.json()["id"] def test_soft_delete_moves_to_trash(self, admin_session, backup_id): r = admin_session.delete(f"{BASE_URL}/api/backup/{backup_id}", timeout=30) assert r.status_code == 200, r.text body = r.json() assert body.get("ok") is True assert body.get("moved_to_trash") is True assert body.get("permanent") is False def test_trash_list_contains_item(self, admin_session, backup_id): r = admin_session.get(f"{BASE_URL}/api/backup/trash", timeout=30) assert r.status_code == 200 items = r.json() item = next((x for x in items if x["id"] == backup_id), None) assert item is not None, "Trashed backup not found in trash list" # Required fields for k in ("id", "filename", "label", "size_bytes", "size_human", "encrypted", "created_at", "deleted_at", "deleted_by", "retention_days", "days_remaining", "restore_path", "total_rows"): assert k in item, f"Missing key {k}" assert item["retention_days"] == 30 assert item["days_remaining"] in (29, 30) assert item["label"].startswith("qa-trash") or "qa" in item["label"] assert "@" in (item["deleted_by"] or "") def test_active_list_excludes_trash(self, admin_session, backup_id): r = admin_session.get(f"{BASE_URL}/api/backup/list", timeout=30) assert r.status_code == 200 ids = [x["id"] for x in r.json()] assert backup_id not in ids, "Soft-deleted backup must NOT appear in active list" def test_auto_clean_within_retention(self, admin_session, backup_id): # Item is fresh (30 day retention) -> purged should be 0 r = admin_session.post(f"{BASE_URL}/api/backup/trash/auto-clean", timeout=30) assert r.status_code == 200 body = r.json() assert body.get("purged") == 0 # Confirm still in trash r2 = admin_session.get(f"{BASE_URL}/api/backup/trash", timeout=30) assert backup_id in [x["id"] for x in r2.json()] def test_restore_from_trash(self, admin_session, backup_id): r = admin_session.post(f"{BASE_URL}/api/backup/trash/{backup_id}/restore", timeout=30) assert r.status_code == 200, r.text body = r.json() assert body.get("ok") is True assert body.get("restored") is True assert "filename" in body # Should re-appear in active list r2 = admin_session.get(f"{BASE_URL}/api/backup/list", timeout=30) ids = [x["id"] for x in r2.json()] assert backup_id in ids # No longer in trash r3 = admin_session.get(f"{BASE_URL}/api/backup/trash", timeout=30) ids_t = [x["id"] for x in r3.json()] assert backup_id not in ids_t def test_permanent_delete_purges(self, admin_session, backup_id): # Hard delete r = admin_session.delete(f"{BASE_URL}/api/backup/{backup_id}?permanent=true", timeout=30) assert r.status_code == 200, r.text body = r.json() assert body.get("ok") is True and body.get("permanent") is True # Gone from both lists r2 = admin_session.get(f"{BASE_URL}/api/backup/list", timeout=30) assert backup_id not in [x["id"] for x in r2.json()] r3 = admin_session.get(f"{BASE_URL}/api/backup/trash", timeout=30) assert backup_id not in [x["id"] for x in r3.json()] def test_trash_empty_purges_all(self, admin_session): # Create a backup, soft-delete it, then empty trash c = admin_session.post(f"{BASE_URL}/api/backup/create", json={"label": "qa-empty-test", "encrypt": False}, timeout=60) assert c.status_code == 200 bk_id = c.json()["id"] d = admin_session.delete(f"{BASE_URL}/api/backup/{bk_id}", timeout=30) assert d.status_code == 200 # Empty trash e = admin_session.post(f"{BASE_URL}/api/backup/trash/empty", timeout=30) assert e.status_code == 200 body = e.json() assert body.get("ok") is True and body.get("purged", 0) >= 1 # Verify trash is empty r = admin_session.get(f"{BASE_URL}/api/backup/trash", timeout=30) assert r.status_code == 200 assert r.json() == [] or all(x["id"] != bk_id for x in r.json()) # ============ Migration — Expenses + Header Aliases ============ def _multipart(session, url, csv_text, entity, filename="test.csv", extra=None): """Helper for multipart upload. Override session Content-Type=application/json by setting None.""" files = {"file": (filename, csv_text.encode("utf-8"), "text/csv")} data = {"entity": entity} if extra: data.update(extra) # Setting Content-Type to None tells requests to drop it so it can set multipart boundary return session.post(url, files=files, data=data, headers={"Content-Type": None}, timeout=30) class TestMigrationExpenses: def test_expenses_preview_all_columns_mapped(self, admin_session): csv_text = "date,category,amount,paid by,description\n2026-01-15,Office Supplies,1500,Cash,Pens and paper\n" r = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/preview", csv_text, "expenses", "expenses.csv") assert r.status_code == 200, r.text body = r.json() assert body["unmapped_columns"] == [] detected = [c.lower() for c in body["detected_columns"]] assert all(c in detected for c in ["date", "category", "amount", "paid by", "description"]) row = body["preview_rows"][0] assert row.get("expense_date") == "2026-01-15" assert row.get("category") == "Office Supplies" assert str(row.get("amount")) == "1500" assert row.get("payment_mode") == "Cash" assert row.get("description") == "Pens and paper" def test_expenses_preview_dd_mm_yyyy_normalised_on_commit(self, admin_session, default_company_id): # Preview just maps; date normalisation happens on commit csv_text = "date,category,amount,paid by,description\n12/06/2026,Travel,500,UPI,Auto fare\n" r = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/preview", csv_text, "expenses") assert r.status_code == 200 # Commit and verify date normalised in DB c = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/commit", csv_text, "expenses", extra={"company_id": default_company_id, "skip_duplicates": "false"}) assert c.status_code == 200, c.text body = c.json() assert body["inserted"] == 1 # Fetch back sample_id = body["sample_ids"][0] # The expense_date should be 2026-06-12 in mongo # Use /api/expenses if available, else use audit list to confirm insertion rx = admin_session.get(f"{BASE_URL}/api/expenses", timeout=30) if rx.status_code == 200: exps = rx.json() inserted = next((e for e in exps if e.get("id") == sample_id or e.get("description") == "Auto fare"), None) if inserted: assert inserted.get("expense_date") == "2026-06-12", f"Date not normalised: {inserted.get('expense_date')}" def test_expenses_commit_inserts_with_metadata(self, admin_session, default_company_id): csv_text = "date,category,amount,paid by,description\n2026-02-01,TEST_v12_3_Travel,1234,Card,TEST_v12_3 cab\n" r = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/commit", csv_text, "expenses", extra={"company_id": default_company_id, "skip_duplicates": "false"}) assert r.status_code == 200, r.text body = r.json() assert body["inserted"] == 1 assert body.get("sample_ids") class TestMigrationAliases: def test_parties_tally_ledger_name_alias(self, admin_session): csv_text = "Ledger Name,phone,city\nTallyParty 1,9999988888,Mumbai\n" r = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/preview", csv_text, "parties") assert r.status_code == 200, r.text body = r.json() assert body["unmapped_columns"] == [], f"Unmapped: {body['unmapped_columns']}" assert body["preview_rows"][0].get("name") == "TallyParty 1" def test_parties_marg_account_name_alias(self, admin_session): csv_text = "Account Name,phone\nMargParty 1,9111122222\n" r = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/preview", csv_text, "parties") assert r.status_code == 200 body = r.json() assert body["unmapped_columns"] == [] assert body["preview_rows"][0].get("name") == "MargParty 1" def test_items_vyapar_marg_aliases(self, admin_session): csv_text = "Item Name,EAN,Max Retail Price,Stock Group,Wholesale Price\nProd 1,8901111222333,500,Apparel,400\n" r = _multipart(admin_session, f"{BASE_URL}/api/data-io/import/preview", csv_text, "items") assert r.status_code == 200, r.text body = r.json() assert body["unmapped_columns"] == [], f"Unmapped: {body['unmapped_columns']}" row = body["preview_rows"][0] assert row.get("barcode") == "8901111222333" assert str(row.get("mrp")) == "500" assert row.get("category") == "Apparel" assert str(row.get("wholesale_price")) == "400" # ============ txn-prefixes archive filter ============ class TestTxnPrefixesArchive: def test_default_excludes_archived(self, admin_session, default_company_id): r = admin_session.get(f"{BASE_URL}/api/txn-prefixes?company_id={default_company_id}&type=sale", timeout=30) assert r.status_code == 200, r.text items = r.json() for p in items: assert not p.get("archived"), f"Archived prefix leaked into default list: {p}" def test_include_archived_returns_all(self, admin_session, default_company_id): r = admin_session.get(f"{BASE_URL}/api/txn-prefixes?company_id={default_company_id}&type=sale&include_archived=true", timeout=30) assert r.status_code == 200 # Just verify response shape — total count >= default all_items = r.json() r2 = admin_session.get(f"{BASE_URL}/api/txn-prefixes?company_id={default_company_id}&type=sale", timeout=30) assert len(all_items) >= len(r2.json())