From a834dc0e9d9a63d0a2919a1f5eced0f08b7e269c Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Tue, 30 Jun 2026 12:14:40 +0530 Subject: [PATCH] seed for new --- scripts/seed_uat_fy_lock_backup_strict.py | 562 ++++++++++++++++++++++ 1 file changed, 562 insertions(+) create mode 100644 scripts/seed_uat_fy_lock_backup_strict.py diff --git a/scripts/seed_uat_fy_lock_backup_strict.py b/scripts/seed_uat_fy_lock_backup_strict.py new file mode 100644 index 0000000..4080a92 --- /dev/null +++ b/scripts/seed_uat_fy_lock_backup_strict.py @@ -0,0 +1,562 @@ +""" +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()