""" Audit Firm ERP -- strict UAT seed patch for FY lock/backup Playwright tests. Run inside ERP container after core UAT seed: cd /app PYTHONPATH=/app python scripts/seed_uat_fy_lock_backup_strict.py Purpose: Creates deterministic UAT records needed by tests/fy-lock-backup.spec.js without weakening test assertions or changing ERP features. Scope: - Financial years: 2025-26 current/open and 2024-25 locked for Tenant A - Tenant/branch B scenario rows where base seed exists - Active FY and locked FY subscriptions/tasks - Engagement document and permanent document records - Notice case record - Billing invoice/payment records The script is idempotent and only upserts UAT-coded records. Intended for UAT/test containers, not live production data. """ from __future__ import annotations import hashlib import json import os import sys from datetime import date, datetime, timedelta, timezone ACTIVE_FY = os.getenv("ACTIVE_FY", "2025-26") LOCKED_FY = os.getenv("LOCKED_FY", "2024-25") ACTIVE_AY = os.getenv("ACTIVE_AY", "2026-27") LOCKED_AY = os.getenv("LOCKED_AY", "2025-26") UAT_DOMAIN = os.getenv("UAT_EMAIL_DOMAIN", "vavalam.com") def fail(msg: str) -> None: print(f"\n[fy-strict-seed] ERROR: {msg}", file=sys.stderr) sys.exit(1) def utcnow() -> datetime: return datetime.now(timezone.utc) def main() -> None: try: from sqlalchemy import MetaData, and_, inspect, select, update, text from app.core.db.common import CommonSessionLocal except Exception as exc: fail(f"Run this from /app inside ERP container. Import error: {exc!r}") db = CommonSessionLocal() engine = db.get_bind() meta = MetaData() meta.reflect(bind=engine) inspector = inspect(engine) table_names = set(inspector.get_table_names()) notes: list[str] = [] def norm(name: str) -> str: return name.lower().replace("_", "").replace("-", "") def table(*names: str): for name in names: if name in meta.tables: return meta.tables[name] lookup = {norm(t): t for t in table_names} for name in names: hit = lookup.get(norm(name)) if hit: return meta.tables[hit] return None def valid(t, data: dict) -> dict: if t is None: return {} cols = set(t.c.keys()) return {k: v for k, v in data.items() if k in cols and v is not None} def where(t, lookup: dict): lookup = valid(t, lookup) if not lookup: return None return and_(*[t.c[k] == v for k, v in lookup.items()]) def one(t, lookup: dict): if t is None: return None clause = where(t, lookup) if clause is None: return None return db.execute(select(t).where(clause)).mappings().first() def first(t, *lookups: dict): if t is None: return None for lookup in lookups: row = one(t, lookup) if row: return row return db.execute(select(t).limit(1)).mappings().first() def scalar(sql: str): try: return db.execute(text(sql)).scalar() except Exception: return None def py_default(t, col_name: str): name = col_name.lower() if name.endswith("_at_utc") or name.endswith("_at"): return utcnow() if name.endswith("_date"): return date.today() if name in {"status", "export_status"}: return "active" if name in {"is_active"}: return True if name in {"is_deleted", "is_locked", "is_current"}: return False try: pytype = t.c[col_name].type.python_type except Exception: pytype = str if pytype is bool: return False if pytype is int: return 0 if pytype is float: return 0.0 if pytype is date: return date.today() if pytype is datetime: return utcnow() return f"UAT {col_name.replace('_', ' ').title()}" def add_required_defaults(t, data: dict) -> dict: out = dict(data) for col in t.c: if col.name in out: continue if col.primary_key or col.nullable or col.default is not None or col.server_default is not None: continue out[col.name] = py_default(t, col.name) return valid(t, out) def upsert(t, lookup: dict, defaults: dict, label: str): if t is None: notes.append(f"SKIP {label}: table missing") return None lookup_v = valid(t, lookup) if not lookup_v: notes.append(f"SKIP {label}: lookup columns missing in {t.name}") return None row = one(t, lookup_v) values = valid(t, defaults) if row: if values: db.execute(update(t).where(where(t, lookup_v)).values(**values)) db.flush() return one(t, lookup_v) data = add_required_defaults(t, {**lookup_v, **values}) try: result = db.execute(t.insert().values(**data)) db.flush() pk = result.inserted_primary_key[0] if result.inserted_primary_key else None if pk and "id" in t.c: return one(t, {"id": pk}) return one(t, lookup_v) except Exception as exc: db.rollback() notes.append(f"SKIP {label}: insert failed in {t.name}: {exc.__class__.__name__}: {str(exc).splitlines()[0]}") return one(t, lookup_v) def v(row, key: str, default=None): return row[key] if row is not None and key in row else default tenants = table("tenants") branches = table("branches") users = table("users") clients = table("clients") service_catalogues = table("service_catalogues", "services") financial_years = table("financial_years") subscriptions = table("client_service_subscriptions") tasks = table("client_service_task_instances") engagement_documents = table("engagement_documents") engagement_document_versions = table("engagement_document_versions") permanent_docs = table("permanent_client_documents") permanent_versions = table("permanent_client_document_versions") notice_cases = table("notice_cases") notice_docs = table("notice_case_documents") invoices = table("billing_invoices") payments = table("billing_payments") tenant_a = first(tenants, {"code": "UAT-A"}, {"tenant_code": "UAT-A"}, {"name": "UAT Tenant A"}) tenant_b = first(tenants, {"code": "UAT-B"}, {"tenant_code": "UAT-B"}, {"name": "UAT Tenant B"}) or tenant_a branch_a = first(branches, {"code": "UAT-BA"}, {"branch_code": "UAT-BA"}, {"name": "UAT Branch A"}) branch_b = first(branches, {"code": "UAT-BB"}, {"branch_code": "UAT-BB"}, {"name": "UAT Branch B"}) or branch_a admin = first(users, {"email": f"uat.admin@{UAT_DOMAIN}"}, {"email": f"uat.firmadmin@{UAT_DOMAIN}"}) firm_admin = first(users, {"email": f"uat.firmadmin@{UAT_DOMAIN}"}) or admin partner = first(users, {"email": f"uat.partner@{UAT_DOMAIN}"}) or firm_admin manager = first(users, {"email": f"uat.manager@{UAT_DOMAIN}"}) or firm_admin staff = first(users, {"email": f"uat.staff@{UAT_DOMAIN}"}) or firm_admin branch_b_staff = first(users, {"email": f"uat.branch2staff@{UAT_DOMAIN}"}) or staff client_user = first(users, {"email": f"uat.client@{UAT_DOMAIN}"}) if not tenant_a or not branch_a or not admin: fail("Core UAT seed missing: tenant A, branch A, or admin user not found.") tenant_a_id = v(tenant_a, "id") tenant_b_id = v(tenant_b, "id") branch_a_id = v(branch_a, "id") branch_b_id = v(branch_b, "id") admin_id = v(admin, "id") partner_id = v(partner, "id") manager_id = v(manager, "id") staff_id = v(staff, "id") branch_b_staff_id = v(branch_b_staff, "id") # Ensure FY records required by the suite. fy_active = upsert(financial_years, {"tenant_id": tenant_a_id, "year_code": ACTIVE_FY}, { "assessment_year": ACTIVE_AY, "start_date": date(int(ACTIVE_FY[:4]), 4, 1), "end_date": date(int(ACTIVE_FY[:4]) + 1, 3, 31), "is_current": True, "is_locked": False, "locked_at_utc": None, "locked_by_user_id": None, "updated_at_utc": utcnow(), }, "active financial year") fy_locked = upsert(financial_years, {"tenant_id": tenant_a_id, "year_code": LOCKED_FY}, { "assessment_year": LOCKED_AY, "start_date": date(int(LOCKED_FY[:4]), 4, 1), "end_date": date(int(LOCKED_FY[:4]) + 1, 3, 31), "is_current": False, "is_locked": True, "locked_at_utc": utcnow(), "locked_by_user_id": admin_id, "updated_at_utc": utcnow(), }, "locked financial year") if financial_years is not None and fy_active: try: db.execute(update(financial_years).where( and_(financial_years.c.tenant_id == tenant_a_id, financial_years.c.year_code != ACTIVE_FY) ).values(is_current=False)) db.execute(update(financial_years).where( and_(financial_years.c.tenant_id == tenant_a_id, financial_years.c.year_code == ACTIVE_FY) ).values(is_current=True, is_locked=False, locked_at_utc=None, locked_by_user_id=None)) db.flush() except Exception as exc: notes.append(f"Could not normalize current FY flag: {exc.__class__.__name__}: {exc}") service = first(service_catalogues, {"service_code": "GST-GSTR3B-M"}, {"code": "GST-GSTR3B-M"}) or first(service_catalogues, {}) if not service: fail("No service catalogue/service row available. Run service seed first.") service_id = v(service, "id") # Ensure Client A and Client B records where possible. client_a = first(clients, {"client_code": "UAT-CL-A"}, {"code": "UAT-CL-A"}, {"client_name": "UAT Client A Pvt Ltd"}) client_a = client_a or upsert(clients, {"tenant_id": tenant_a_id, "client_code": "UAT-CL-A"}, { "branch_id": branch_a_id, "client_name": "UAT Client A Pvt Ltd", "trade_name": "UAT Client A", "pan": "ABCDE1234F", "gstin": "33ABCDE1234F1Z5", "email": f"uat.client@{UAT_DOMAIN}", "mobile": "9000000004", "status": "active", "is_active": True, "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, "client A") client_b = first(clients, {"client_code": "UAT-CL-B"}, {"code": "UAT-CL-B"}, {"client_name": "UAT Client B Pvt Ltd"}) client_b = client_b or upsert(clients, {"tenant_id": tenant_b_id, "client_code": "UAT-CL-B"}, { "branch_id": branch_b_id, "client_name": "UAT Client B Pvt Ltd", "trade_name": "UAT Client B", "pan": "ABCDE5678F", "gstin": "33ABCDE5678F1Z5", "email": f"uat.client2@{UAT_DOMAIN}", "mobile": "9000000005", "status": "active", "is_active": True, "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, "client B") client_a_id = v(client_a, "id") client_b_id = v(client_b, "id", client_a_id) # Subscriptions / engagements. def sub_seed(label, tenant_id, branch_id, client_id, fy, ay, assignee_staff_id, locked=False): return upsert(subscriptions, {"tenant_id": tenant_id, "client_id": client_id, "service_catalogue_id": service_id, "financial_year": fy}, { "branch_id": branch_id, "assessment_year": ay, "assigned_partner_user_id": partner_id, "assigned_manager_user_id": manager_id, "assigned_staff_user_id": assignee_staff_id, "review_partner_user_id": partner_id, "engagement_type": "non_assurance", "status": "active", "is_active": True, "is_locked": locked, "locked_at_utc": utcnow() if locked else None, "locked_by_user_id": admin_id if locked else None, "start_date": date(int(fy[:4]), 4, 1), "end_date": date(int(fy[:4]) + 1, 3, 31), "original_due_date": date(int(fy[:4]) + 1, 4, 20), "current_due_date": date(int(fy[:4]) + 1, 4, 20), "remarks": f"Strict UAT seed engagement for {label}", "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, label) sub_active = sub_seed("active FY subscription", tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, staff_id, False) sub_locked = sub_seed("locked FY subscription", tenant_a_id, branch_a_id, client_a_id, LOCKED_FY, LOCKED_AY, staff_id, True) sub_branch_b = sub_seed("branch B subscription", tenant_b_id, branch_b_id, client_b_id, ACTIVE_FY, ACTIVE_AY, branch_b_staff_id, False) # Task instances. def task_seed(label, sub, tenant_id, branch_id, client_id, fy, ay, task_name, assignee_id, status="pending", locked=False, seq=1): return upsert(tasks, {"subscription_id": v(sub, "id"), "financial_year": fy, "task_name": task_name}, { "tenant_id": tenant_id, "branch_id": branch_id, "client_id": client_id, "service_catalogue_id": service_id, "assessment_year": ay, "description": f"Strict UAT seed task for {label}", "sequence_no": seq, "default_role_name": "Staff", "assigned_to_user_id": assignee_id, "internal_target_date": date.today() + timedelta(days=10), "status": status, "priority": "normal", "is_active": True, "is_locked": locked, "locked_at_utc": utcnow() if locked else None, "locked_by_user_id": admin_id if locked else None, "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, label) task_active = task_seed("active FY task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT Active FY Task", staff_id, "pending", False, 10) task_partner = task_seed("partner review task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT Partner Review Task", partner_id, "pending", False, 20) task_in_progress = task_seed("in progress task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT In Progress Task", staff_id, "in_progress", False, 30) task_completed = task_seed("completed task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT Completed Task", staff_id, "completed", False, 40) task_locked = task_seed("locked FY task", sub_locked, tenant_a_id, branch_a_id, client_a_id, LOCKED_FY, LOCKED_AY, "UAT Locked FY Task", staff_id, "pending", True, 50) task_branch_b = task_seed("branch B active FY task", sub_branch_b, tenant_b_id, branch_b_id, client_b_id, ACTIVE_FY, ACTIVE_AY, "UAT Branch B Active FY Task", branch_b_staff_id, "pending", False, 60) # Engagement document and version. doc_content = b"Strict UAT FY engagement document\n" eng_doc = upsert(engagement_documents, {"tenant_id": tenant_a_id, "engagement_id": v(sub_active, "id"), "document_code": "UAT-FY-ENG-DOC-001"}, { "branch_id": branch_a_id, "client_id": client_a_id, "task_instance_id": v(task_active, "id"), "financial_year": ACTIVE_FY, "assessment_year": ACTIVE_AY, "document_type": "GENERAL", "title": "UAT FY Engagement Document", "description": "Strict UAT seed engagement document", "current_version_no": 1, "status": "active", "is_deleted": False, "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, "engagement document") eng_doc_ver = upsert(engagement_document_versions, {"document_id": v(eng_doc, "id"), "version_no": 1}, { "tenant_id": tenant_a_id, "branch_id": branch_a_id, "client_id": client_a_id, "engagement_id": v(sub_active, "id"), "original_filename": "uat-fy-engagement-document.txt", "stored_filename": "uat-fy-engagement-document.txt", "content_type": "text/plain", "file_size_bytes": len(doc_content), "file_hash_sha256": hashlib.sha256(doc_content).hexdigest(), "storage_backend": "LOCAL_YEAR_WISE", "local_relative_path": f"UAT-A/{ACTIVE_FY}/engagements/uat-fy-engagement-document.txt", "storage_status": "stored", "remarks": "Strict UAT seed engagement document version", "uploaded_by_user_id": admin_id, "uploaded_at_utc": utcnow(), }, "engagement document version") # Permanent document. perm_doc = upsert(permanent_docs, {"tenant_id": tenant_a_id, "client_id": client_a_id, "document_code": "UAT-FY-PERM-DOC-001"}, { "branch_id": branch_a_id, "category": "KYC", "title": "UAT FY Permanent Client Document", "description": "Strict UAT seed permanent document", "current_version_no": 1, "status": "active", "is_deleted": False, "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, "permanent client document") perm_content = b"Strict UAT permanent document\n" perm_doc_ver = upsert(permanent_versions, {"document_id": v(perm_doc, "id"), "version_no": 1}, { "tenant_id": tenant_a_id, "branch_id": branch_a_id, "client_id": client_a_id, "original_filename": "uat-fy-permanent-document.txt", "stored_filename": "uat-fy-permanent-document.txt", "content_type": "text/plain", "file_size_bytes": len(perm_content), "file_hash_sha256": hashlib.sha256(perm_content).hexdigest(), "storage_backend": "LOCAL_PERMANENT", "local_relative_path": "UAT-A/UAT-CL-A/permanent/uat-fy-permanent-document.txt", "storage_status": "stored", "remarks": "Strict UAT seed permanent document version", "uploaded_by_user_id": admin_id, "uploaded_at_utc": utcnow(), }, "permanent client document version") # Notice case and document. notice_case = upsert(notice_cases, {"tenant_id": tenant_a_id, "case_code": "UAT-FY-NC-001"}, { "branch_id": branch_a_id, "client_id": client_a_id, "engagement_id": v(sub_active, "id"), "service_catalogue_id": service_id, "financial_year": ACTIVE_FY, "assessment_year": ACTIVE_AY, "reference_no": "UAT-FY-NOTICE-001", "notice_no": "UAT-FY-NOTICE-001", "case_title": "UAT FY Notice Case", "title": "UAT FY Notice Case", "department": "GST", "case_type": "notice", "status": "open", "priority": "normal", "notice_date": date.today(), "due_date": date.today() + timedelta(days=30), "description": "Strict UAT seed notice case", "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), "is_active": True, }, "notice case") notice_doc = upsert(notice_docs, {"tenant_id": tenant_a_id, "case_id": v(notice_case, "id"), "title": "UAT FY Notice Case Document"}, { "branch_id": branch_a_id, "document_type": "GENERAL", "description": "Strict UAT seed notice case document", "original_filename": "uat-fy-notice-document.txt", "stored_filename": "uat-fy-notice-document.txt", "content_type": "text/plain", "file_size_bytes": 29, "local_relative_path": "notice_cases/uat-fy-notice-document.txt", "version_no": 1, "status": "active", "is_deleted": False, "uploaded_by_user_id": admin_id, "uploaded_at_utc": utcnow(), }, "notice case document") # Billing invoice and payment. invoice = upsert(invoices, {"tenant_id": tenant_a_id, "invoice_no": f"UAT-FY-INV-{ACTIVE_FY}"}, { "branch_id": branch_a_id, "client_id": client_a_id, "financial_year": ACTIVE_FY, "invoice_date": date.today(), "billing_period_from": date(int(ACTIVE_FY[:4]), 4, 1), "billing_period_to": date(int(ACTIVE_FY[:4]) + 1, 3, 31), "status": "ISSUED", "subtotal_amount": 10000.00, "taxable_amount": 10000.00, "cgst_amount": 900.00, "sgst_amount": 900.00, "igst_amount": 0.00, "total_tax_amount": 1800.00, "total_amount": 11800.00, "amount_paid": 1000.00, "balance_amount": 10800.00, "notes": "Strict UAT seed invoice for FY tests", "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, "billing invoice") payment = upsert(payments, {"tenant_id": tenant_a_id, "receipt_no": f"UAT-FY-RCPT-{ACTIVE_FY}"}, { "branch_id": branch_a_id, "client_id": client_a_id, "invoice_id": v(invoice, "id"), "financial_year": ACTIVE_FY, "payment_date": date.today(), "amount_received": 1000.00, "tds_deducted": 0.00, "mode": "BANK_TRANSFER", "reference_no": "UAT-FY-SEED", "notes": "Strict UAT seed payment for FY tests", "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), }, "billing payment") db.commit() env_ids = { "TENANT_A_ID": tenant_a_id, "TENANT_B_ID": tenant_b_id, "BRANCH_A_ID": branch_a_id, "BRANCH_B_ID": branch_b_id, "CLIENT_A_ID": client_a_id, "CLIENT_B_ID": client_b_id, "ENGAGEMENT_A_ID": v(sub_active, "id"), "ENGAGEMENT_B_ID": v(sub_branch_b, "id"), "SUBSCRIPTION_A_ID": v(sub_active, "id"), "ACTIVE_FY_TASK_ID": v(task_active, "id"), "TASK_A_ID": v(task_active, "id"), "LOCKED_FY_TASK_ID": v(task_locked, "id"), "PARTNER_TASK_A_ID": v(task_partner, "id"), "BRANCH_B_TASK_ID": v(task_branch_b, "id"), "TASK_DOCUMENT_A_ID": v(eng_doc, "id"), "ENGAGEMENT_DOCUMENT_A_ID": v(eng_doc, "id"), "PERM_DOCUMENT_ID": v(perm_doc, "id"), "NOTICE_CASE_A_ID": v(notice_case, "id"), "CASE_DOCUMENT_A_ID": v(notice_doc, "id"), "INVOICE_A_ID": v(invoice, "id"), "PAYMENT_A_ID": v(payment, "id"), "ACTIVE_FY": ACTIVE_FY, "LOCKED_FY": LOCKED_FY, "DEFAULT_YEAR_CODE": ACTIVE_FY, "WORK_TRACKER_ROUTE": "/services/work-tracker", "WORK_TRACKER_FALLBACK_ROUTE": "/employee/work", "DOCUMENTS_ROUTE": "/documents", "PERMANENT_DOCUMENTS_ROUTE": "/documents/permanent", "NOTICE_CASES_ROUTE": "/notice-cases", "BILLING_ROUTE": "/billing", "CLIENT_BILLING_ROUTE": "/client/billing", } print("\n[fy-strict-seed] Completed strict FY lock/backup seed patch.") print("\nCopy/update these in /tests/.env if different:\n") for k, val in env_ids.items(): print(f"{k}={val or ''}") if notes: print("\nNotes / non-fatal skips:") for note in notes: print(f"- {note}") else: print("\nNotes: no non-fatal seed skips reported.") db.close() if __name__ == "__main__": main()