From a96c1ba49b5f335a8f4a1cf2a0904352587e06da Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Wed, 24 Jun 2026 21:58:50 +0530 Subject: [PATCH] Fix v2.5.1 seed platform billing account fields --- scripts/seed_uat_data_v2_5_1_FIX3.py | 650 +++++++++++++++++++++++++++ 1 file changed, 650 insertions(+) create mode 100644 scripts/seed_uat_data_v2_5_1_FIX3.py diff --git a/scripts/seed_uat_data_v2_5_1_FIX3.py b/scripts/seed_uat_data_v2_5_1_FIX3.py new file mode 100644 index 0000000..416bb55 --- /dev/null +++ b/scripts/seed_uat_data_v2_5_1_FIX3.py @@ -0,0 +1,650 @@ +""" +Audit Firm ERP -- UAT seed extension for Playwright v2.5.1 Deep SQLite suite. + +Purpose: + Extend the existing core UAT seed with supporting records for the new v2.5.1 + deep tests: email integration, marketplace leads, platform billing, RBAC, + audit logs, document security, HR, billing rules, notice-case rules and + financial-year locking. + +Run inside ERP container, AFTER the core seed: + + cd /app + PYTHONPATH=/app python scripts/seed_uat_data.py + PYTHONPATH=/app python scripts/seed_uat_data_v2_5_1.py + +Safe/idempotent for UAT: it inserts or updates only UAT-coded records where the +matching tables/columns exist. If a module table is not present in this ERP build, +the script skips that area and prints a warning instead of failing. + +Do NOT run against live production data. +""" +from __future__ import annotations + +import hashlib +import json +import os +import sys +from datetime import date, datetime, timedelta, timezone + +UAT_DOMAIN = os.getenv("UAT_EMAIL_DOMAIN", "vavalam.com") +ACTIVE_FY = os.getenv("ACTIVE_FY", "2025-26") +PREVIOUS_FY = os.getenv("PREVIOUS_FY", "2024-25") +ASSESSMENT_YEAR = os.getenv("ASSESSMENT_YEAR", "2026-27") +LSA_SECRET = os.getenv("UAT_LSA_SECRET", "uat-local-storage-agent-secret") +SMTP_PASSWORD = os.getenv("UAT_SMTP_PASSWORD", "Pass@123##") + + +def fail(msg: str) -> None: + print(f"\n[v2.5.1 seed] ERROR: {msg}", file=sys.stderr) + sys.exit(1) + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def today() -> date: + return date.today() + + +def main() -> None: + try: + from sqlalchemy import MetaData, and_, inspect, select, update + from app.core.db.common import CommonSessionLocal + except Exception as exc: + fail( + "Could not import ERP DB modules. Run this from ERP project root inside the app container. " + f"Underlying 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()) + + report: list[str] = [] + ids: dict[str, object] = {} + + def norm(name: str) -> str: + return name.lower().replace("_", "").replace("-", "") + + def find_table(*candidates: str): + # exact match first + for name in candidates: + if name in meta.tables: + return meta.tables[name] + # normalized exact match + normalized = {norm(t): t for t in table_names} + for name in candidates: + hit = normalized.get(norm(name)) + if hit: + return meta.tables[hit] + # token contains fallback + for name in candidates: + tokens = [tok for tok in name.lower().replace("-", "_").split("_") if tok] + for t in table_names: + low = t.lower() + if all(tok in low for tok in tokens): + return meta.tables[t] + return None + + def valid(table, data: dict) -> dict: + if table is None: + return {} + cols = set(table.c.keys()) + return {k: v for k, v in data.items() if k in cols} + + def required_default_for_column(table, column_name: str): + """Best-effort UAT defaults for reflected NOT NULL columns. + + This keeps the seed resilient when the ERP schema has extra mandatory + fields that are not present in every build/version. Values are only + added when the column exists and is NOT NULL with no server/default. + """ + name = column_name.lower() + table_name = table.name.lower() if table is not None else "" + + # Common foreign-key/context IDs + if name in {"tenant_id", "audit_firm_id"}: + return tenant_a_id + if name == "branch_id": + return branch_a_id + if name in {"created_by_user_id", "updated_by_user_id", "created_by_id", "updated_by_id"}: + return admin_id + if name in {"user_id", "employee_user_id"}: + return staff_id + if name == "client_id": + return client_a_id + if name in {"assigned_to_user_id", "owner_user_id"}: + return manager_id + + # Platform billing mandatory classifiers + if name == "account_type": + return "audit_firm" + if name == "plan_type": + return "audit_firm" + if name == "subscription_type": + return "audit_firm" + if name == "invoice_type": + return "subscription" + if name == "payment_type": + return "invoice" + if name == "entity_type": + return "audit_firm" + + # Email template mandatory columns + if name in {"body_template", "plain_body", "html_template", "body", "content"}: + return "Dear User, your UAT OTP is {{otp}}. This is for ERP UAT testing." + if name in {"subject_template", "subject"}: + return "UAT ERP Test Notification" + if name in {"template_name", "name", "title", "account_name", "display_name"}: + return f"UAT {table_name.replace('_', ' ').title()}" + if name == "template_type": + return "system" + + # Common business defaults + if name == "currency": + return "INR" + if name == "billing_cycle": + return "monthly" + if name == "status": + return "active" + if name == "gst_registration_type": + return "regular" + if name in {"state", "place_of_supply"}: + return "Tamil Nadu" + if name in {"country"}: + return "India" + if name in {"address", "billing_address"}: + return "UAT Test Address" + if name in {"email", "billing_email"}: + return "support@vavalam.com" + if name in {"mobile", "phone", "contact_number"}: + return "9000000000" + if name.endswith("_code") or name in {"code", "account_code", "invoice_no", "payment_ref"}: + return f"UAT-{table_name[:10].upper()}" + if name.endswith("_no") or name.endswith("_number"): + return f"UAT-{table_name[:8].upper()}-001" + if name in {"description", "remarks", "notes"}: + return "Seeded for Playwright v2.5.1 UAT/VAPT testing" + + # Date/time defaults + if name.endswith("_date") or name in {"start_date", "end_date", "due_date", "invoice_date", "payment_date"}: + return today() + if name.endswith("_at") or name in {"created_at", "updated_at"}: + return utcnow() + + # Type based fallbacks, avoiding unknown FK IDs where possible + try: + pytype = table.c[column_name].type.python_type + except Exception: + pytype = str + if pytype is bool: + return False + if pytype is int: + if name.endswith("_id"): + return None + return 0 + if pytype is float: + return 0.0 + if pytype is date: + return today() + if pytype is datetime: + return utcnow() + return f"UAT {column_name.replace('_', ' ').title()}" + + def add_required_defaults(table, data: dict) -> dict: + if table is None: + return data + out = dict(data) + for col in table.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 + value = required_default_for_column(table, col.name) + if value is not None: + out[col.name] = value + return valid(table, out) + + def one(table, lookup: dict): + if table is None: + return None + data = valid(table, lookup) + if not data: + return None + stmt = select(table).where(and_(*[table.c[k] == v for k, v in data.items()])) + return db.execute(stmt).mappings().first() + + def first(table, **lookup): + return one(table, lookup) + + def first_by_any(table, lookups: list[dict]): + for lookup in lookups: + row = one(table, lookup) + if row: + return row + return None + + def upsert(table, lookup: dict, defaults: dict | None = None, label: str = "record"): + if table is None: + report.append(f"SKIP {label}: table missing") + return None + lookup_v = valid(table, lookup) + if not lookup_v: + report.append(f"SKIP {label}: no matching lookup columns in {table.name}") + return None + defaults_v = valid(table, defaults or {}) + row = one(table, lookup_v) + if row: + if defaults_v: + db.execute(update(table).where(and_(*[table.c[k] == v for k, v in lookup_v.items()])).values(**defaults_v)) + db.flush() + return one(table, lookup_v) + data = add_required_defaults(table, {**lookup_v, **defaults_v}) + result = db.execute(table.insert().values(**data)) + db.flush() + pk = result.inserted_primary_key[0] if result.inserted_primary_key else None + if pk and "id" in table.c: + return one(table, {"id": pk}) + return one(table, lookup_v) + + def update_existing(table, lookup: dict, values: dict, label: str): + if table is None: + report.append(f"SKIP {label}: table missing") + return None + lookup_v = valid(table, lookup) + values_v = valid(table, values) + if not lookup_v or not values_v: + report.append(f"SKIP {label}: no matching columns in {table.name}") + return None + db.execute(update(table).where(and_(*[table.c[k] == v for k, v in lookup_v.items()])).values(**values_v)) + db.flush() + return one(table, lookup_v) + + def val(row, key: str, default=None): + return row[key] if row and key in row else default + + # Core table references from earlier seed + tenants = find_table("tenants", "tenant") + branches = find_table("branches", "branch") + users = find_table("users", "user") + clients = find_table("clients", "client") + employees = find_table("employees", "employee") + financial_years = find_table("financial_years", "financial_year") + documents = find_table("engagement_documents", "documents", "client_documents") + document_versions = find_table("engagement_document_versions", "document_versions") + storage_nodes = find_table("branch_storage_nodes", "storage_nodes") + storage_jobs = find_table("document_storage_jobs", "storage_jobs") + notice_cases = find_table("notice_cases", "cases") + subscriptions = find_table("client_service_subscriptions", "subscriptions", "engagements") + task_instances = find_table("client_service_task_instances", "task_instances", "tasks") + services = find_table("service_catalogue", "service_catalogues", "services") + roles = find_table("roles", "rbac_roles") + permissions = find_table("permissions", "rbac_permissions") + role_permissions = find_table("role_permissions", "rbac_role_permissions") + + tenant_a = first_by_any(tenants, [{"code": "UAT-A"}, {"tenant_code": "UAT-A"}, {"name": "UAT Tenant A"}]) + tenant_b = first_by_any(tenants, [{"code": "UAT-B"}, {"tenant_code": "UAT-B"}, {"name": "UAT Tenant B"}]) + branch_a = first_by_any(branches, [{"code": "UAT-BA"}, {"branch_code": "UAT-BA"}, {"name": "UAT Branch A"}]) + branch_b = first_by_any(branches, [{"code": "UAT-BB"}, {"branch_code": "UAT-BB"}, {"name": "UAT Branch B"}]) + firm_admin = first(users, email=f"uat.firmadmin@{UAT_DOMAIN}") + system_admin = first(users, email=f"uat.admin@{UAT_DOMAIN}") + partner = first(users, email=f"uat.partner@{UAT_DOMAIN}") + manager = first(users, email=f"uat.manager@{UAT_DOMAIN}") + staff = first(users, email=f"uat.staff@{UAT_DOMAIN}") + staff2 = first(users, email=f"uat.staff2@{UAT_DOMAIN}") + client_user = first(users, email=f"uat.client@{UAT_DOMAIN}") + consultant = first(users, email=f"uat.consultant@{UAT_DOMAIN}") + client_a = first_by_any(clients, [{"client_code": "UAT-CL-A"}, {"code": "UAT-CL-A"}, {"client_name": "UAT Client A Pvt Ltd"}]) + client_b = first_by_any(clients, [{"client_code": "UAT-CL-B"}, {"code": "UAT-CL-B"}, {"client_name": "UAT Client B Pvt Ltd"}]) + service_a = first_by_any(services, [{"service_code": "GST-GSTR3B-M"}, {"code": "GST-GSTR3B-M"}, {"service_name": "GSTR-3B Monthly Filing"}]) + sub_a = first_by_any(subscriptions, [{"financial_year": ACTIVE_FY, "client_id": val(client_a, "id")}, {"client_id": val(client_a, "id")}]) + task_a = first_by_any(task_instances, [{"client_id": val(client_a, "id")}, {"assigned_to_user_id": val(staff, "id")}]) + + if not tenant_a or not branch_a or not firm_admin: + fail("Core UAT seed appears missing. Run seed_uat_data.py first.") + + tenant_a_id = val(tenant_a, "id") + tenant_b_id = val(tenant_b, "id") + branch_a_id = val(branch_a, "id") + branch_b_id = val(branch_b, "id") + firm_admin_id = val(firm_admin, "id") + system_admin_id = val(system_admin, "id", firm_admin_id) + partner_id = val(partner, "id", firm_admin_id) + manager_id = val(manager, "id", firm_admin_id) + staff_id = val(staff, "id", firm_admin_id) + staff2_id = val(staff2, "id", staff_id) + client_user_id = val(client_user, "id", firm_admin_id) + consultant_id = val(consultant, "id", firm_admin_id) + client_a_id = val(client_a, "id") + client_b_id = val(client_b, "id") + service_a_id = val(service_a, "id") + sub_a_id = val(sub_a, "id") + task_a_id = val(task_a, "id") + + common_scope = { + "tenant_id": tenant_a_id, + "branch_id": branch_a_id, + "created_by_user_id": firm_admin_id, + "updated_by_user_id": firm_admin_id, + "created_by_id": firm_admin_id, + "updated_by_id": firm_admin_id, + "created_at": utcnow(), + "updated_at": utcnow(), + "is_active": True, + } + + # 1. Email integration seed + email_settings = find_table("email_settings", "firm_email_settings", "smtp_settings", "mail_settings") + smtp = upsert(email_settings, {"tenant_id": tenant_a_id, "setting_code": "UAT-SMTP-A"}, { + **common_scope, + "name": "UAT SMTP Settings", + "smtp_host": "mail.vavalam.com", + "smtp_port": 587, + "smtp_username": "no-reply@vavalam.com", + "smtp_password": SMTP_PASSWORD, + "from_email": "no-reply@vavalam.com", + "from_name": "ARRR ERP UAT", + "use_tls": True, + "imap_host": "mail.vavalam.com", + "imap_port": 993, + "imap_username": f"uat.firmadmin@{UAT_DOMAIN}", + "imap_password": SMTP_PASSWORD, + "status": "active", + "is_default": True, + }, "email settings") + ids["EMAIL_SETTINGS_ID"] = val(smtp, "id") + + email_templates = find_table("email_templates", "mail_templates") + for code, subject, body in [ + ("UAT_OTP", "Your ERP OTP", "Your OTP is {{ otp }}"), + ("UAT_INVITE", "ERP Invitation", "Please accept invitation: {{ invite_url }}"), + ("UAT_NOTICE", "Notice update", "Notice {{ reference_no }} requires attention"), + ]: + template_label = code.replace("_", " ") + upsert(email_templates, {"tenant_id": tenant_a_id, "template_code": code}, { + **common_scope, + # Different ERP builds use different mandatory/template column names. + "name": template_label, + "template_name": template_label, + "title": template_label, + "subject": subject, + "email_subject": subject, + "subject_template": subject, + "template_subject": subject, + "body": body, + "body_text": body, + "body_html": body, + "html_body": body, + "template_body": body, + "body_template": body, + "plain_body": body, + "html_template": body, + "content": body, + "template_content": body, + "template_type": "system", + "category": "system", + "is_active": True, + }, f"email template {code}") + + email_queue = find_table("email_queue", "queued_emails", "mail_queue") + upsert(email_queue, {"tenant_id": tenant_a_id, "queue_code": "UAT-EMAIL-Q-001"}, { + **common_scope, "to_email": f"uat.client@{UAT_DOMAIN}", "from_email": "no-reply@vavalam.com", + "subject": "UAT queued email", "body": "Queued email for UAT", "status": "pending", + "priority": 5, "scheduled_at": utcnow(), "attempts": 0, + }, "email queue") + + email_logs = find_table("email_logs", "mail_logs", "email_delivery_logs") + upsert(email_logs, {"tenant_id": tenant_a_id, "message_id": "UAT-EMAIL-LOG-001"}, { + **common_scope, "to_email": f"uat.client@{UAT_DOMAIN}", "from_email": "no-reply@vavalam.com", + "subject": "UAT delivered email", "status": "sent", "sent_at": utcnow(), "provider_response": "UAT seeded", + }, "email log") + + inbox = find_table("email_inbox_messages", "email_inbox", "inbound_emails", "mail_inbox") + upsert(inbox, {"tenant_id": tenant_a_id, "message_id": "UAT-INBOX-001"}, { + **common_scope, "from_email": f"uat.client@{UAT_DOMAIN}", "to_email": "support@vavalam.com", + "subject": "UAT client document submission", "body": "Please map this email to the UAT client.", + "received_at": utcnow(), "status": "unmapped", "client_id": client_a_id, + }, "email inbox") + + # 2. Marketplace / public lead seed + leads = find_table("marketplace_leads", "leads", "public_leads") + lead1 = upsert(leads, {"tenant_id": tenant_a_id, "lead_code": "UAT-LEAD-001"}, { + **common_scope, "source": "public_form", "lead_type": "compliance", "name": "UAT Marketplace Lead", + "company_name": "UAT Lead Pvt Ltd", "contact_name": "UAT Lead Contact", "email": "lead@example.com", + "mobile": "9000000101", "service_interest": "GST", "status": "new", "assigned_to_user_id": manager_id, + "notes": "Seeded marketplace lead for v2.5.1 tests", + }, "marketplace lead new") + lead2 = upsert(leads, {"tenant_id": tenant_a_id, "lead_code": "UAT-LEAD-CONVERTED"}, { + **common_scope, "source": "referral", "name": "UAT Converted Lead", "email": "converted@example.com", + "mobile": "9000000102", "status": "converted", "assigned_to_user_id": partner_id, "converted_client_id": client_a_id, + }, "marketplace lead converted") + ids["MARKETPLACE_LEAD_ID"] = val(lead1, "id") + + # 3. Platform billing seed + plans = find_table("platform_billing_plans", "billing_plans", "plans") + plan = upsert(plans, {"plan_code": "UAT-PLAN-PRO"}, { + "name": "UAT Professional Plan", "description": "Seed plan for platform billing tests", + "monthly_price": 9999, "annual_price": 99990, "currency": "INR", "max_users": 25, + "max_clients": 500, "is_active": True, "created_at": utcnow(), "updated_at": utcnow(), + }, "platform plan") + accounts = find_table("platform_billing_accounts", "billing_accounts", "accounts") + account = upsert(accounts, {"tenant_id": tenant_a_id, "account_code": "UAT-PLAT-ACC-A"}, { + **common_scope, "account_type": "audit_firm", "account_name": "UAT Tenant A Billing Account", + "display_name": "UAT Tenant A Billing Account", "billing_email": "support@vavalam.com", + "gstin": "33AABCU1111A1Z5", "gst_registration_type": "regular", + "billing_address": "UAT Test Address", "state": "Tamil Nadu", "status": "active", + }, "platform billing account") + plat_subs = find_table("platform_subscriptions", "platform_billing_subscriptions", "audit_firm_subscriptions", "subscriptions") + plat_sub = upsert(plat_subs, {"tenant_id": tenant_a_id, "subscription_code": "UAT-PLAT-SUB-A"}, { + **common_scope, "account_id": val(account, "id"), "plan_id": val(plan, "id"), "status": "active", + "start_date": today(), "end_date": today() + timedelta(days=365), "billing_cycle": "monthly", + }, "platform subscription") + plat_invoices = find_table("platform_invoices", "platform_billing_invoices", "invoices") + p_inv = upsert(plat_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-PLAT-INV-001"}, { + **common_scope, "account_id": val(account, "id"), "subscription_id": val(plat_sub, "id"), + "invoice_date": today(), "due_date": today() + timedelta(days=15), "status": "issued", + "subtotal": 9999, "tax_amount": 1799.82, "total_amount": 11798.82, "balance_amount": 11798.82, + }, "platform invoice") + payments = find_table("platform_payments", "invoice_payments", "payments") + upsert(payments, {"tenant_id": tenant_a_id, "payment_ref": "UAT-PLAT-PAY-001"}, { + **common_scope, "invoice_id": val(p_inv, "id"), "payment_date": today(), "amount": 5000, + "payment_mode": "bank_transfer", "status": "posted", "remarks": "Seed partial platform payment", + }, "platform payment") + ids["PLATFORM_INVOICE_ID"] = val(p_inv, "id") + + # 4. RBAC seed + perm_codes = [ + "uat.clients.read", "uat.documents.read", "uat.billing.read", "uat.email.manage", "uat.marketplace.manage", + ] + for code in perm_codes: + upsert(permissions, {"code": code}, { + "name": code, "description": f"UAT permission {code}", "module": "UAT", "is_active": True, + "created_at": utcnow(), "updated_at": utcnow(), + }, f"permission {code}") + role = upsert(roles, {"name": "UAT Limited Tester"}, { + "description": "Seeded limited role for RBAC tests", "is_system": False, "is_active": True, + "tenant_id": tenant_a_id, "created_at": utcnow(), "updated_at": utcnow(), + }, "rbac role") + if role_permissions is not None and role is not None: + for code in perm_codes[:2]: + perm = first(permissions, code=code) if permissions is not None else None + upsert(role_permissions, {"role_id": val(role, "id"), "permission_id": val(perm, "id")}, { + "created_at": utcnow(), "created_by_user_id": firm_admin_id, + }, f"role permission {code}") + + # 5. Audit logs seed + audit_logs = find_table("audit_logs", "activity_logs", "system_audit_logs") + for code, action, entity in [ + ("UAT-AUD-LOGIN-FAIL", "login_failed", "User"), + ("UAT-AUD-DOC-DOWNLOAD", "document_download", "EngagementDocument"), + ("UAT-AUD-PERM-DENIED", "permission_denied", "RBAC"), + ]: + upsert(audit_logs, {"tenant_id": tenant_a_id, "event_code": code}, { + **common_scope, "user_id": firm_admin_id, "action": action, "event_type": action, + "entity_type": entity, "entity_id": client_a_id or 1, "ip_address": "127.0.0.1", + "user_agent": "UAT seed", "details": json.dumps({"seed": True, "no_secret": True}), + "message": f"Seed audit event {action}", + }, f"audit log {code}") + + # 6. Financial-year states + update_existing(financial_years, {"tenant_id": tenant_a_id, "year_code": PREVIOUS_FY}, { + "is_locked": True, "locked_at": utcnow(), "locked_by_user_id": firm_admin_id, + "lock_reason": "UAT locked FY for v2.5.1 tests", + }, "previous FY lock") + update_existing(financial_years, {"tenant_id": tenant_a_id, "year_code": ACTIVE_FY}, { + "is_locked": False, "is_current": True, + }, "active FY current/unlocked") + + backups = find_table("financial_year_backups", "fy_backups", "backup_exports") + fy = first(financial_years, tenant_id=tenant_a_id, year_code=PREVIOUS_FY) + backup = upsert(backups, {"tenant_id": tenant_a_id, "backup_code": "UAT-FY-BACKUP-001"}, { + **common_scope, "financial_year_id": val(fy, "id"), "year_code": PREVIOUS_FY, + "status": "completed", "file_name": "uat-fy-backup-001.zip", "file_size_bytes": 1024, + "created_by_user_id": firm_admin_id, + }, "FY backup") + ids["YEAR_BACKUP_EXPORT_ID"] = val(backup, "id") + ids["LOCKED_FY"] = PREVIOUS_FY + + # 7. Document security/deep seed + secret_hash = hashlib.sha256(LSA_SECRET.encode()).hexdigest() + node = upsert(storage_nodes, {"tenant_id": tenant_a_id, "branch_id": branch_a_id, "node_code": "UAT-LSA-SEC-A"}, { + **common_scope, "node_name": "UAT Security Storage Agent", "secret_key_hash": secret_hash, + "storage_root_path": "D:/UAT/Security", "storage_mode": "pull_jobs", "status": "active", + "quota_limit_bytes": 1024 * 1024 * 1024, "used_storage_bytes": 0, + }, "security storage node") + ids["STORAGE_NODE_SECURITY_ID"] = val(node, "id") + + doc_sec = upsert(documents, {"tenant_id": tenant_a_id, "document_code": "UAT-DOC-SEC-001"}, { + **common_scope, "branch_id": branch_a_id, "client_id": client_a_id, "engagement_id": sub_a_id, + "task_instance_id": task_a_id, "financial_year": ACTIVE_FY, "assessment_year": ASSESSMENT_YEAR, + "document_type": "SECURITY", "title": "UAT Client Visible Document", "description": "Client visible UAT doc", + "current_version_no": 2, "status": "active", "is_deleted": False, "is_client_visible": True, + }, "client visible document") + ids["CLIENT_VISIBLE_DOCUMENT_ID"] = val(doc_sec, "id") + for version, filename in [(1, "uat-sec-v1.txt"), (2, "uat-sec-v2.txt")]: + content = f"UAT secure version {version}\n".encode() + upsert(document_versions, {"document_id": val(doc_sec, "id"), "version_no": version}, { + **common_scope, "tenant_id": tenant_a_id, "branch_id": branch_a_id, "client_id": client_a_id, + "engagement_id": sub_a_id, "original_filename": filename, "stored_filename": filename, + "content_type": "text/plain", "file_size_bytes": len(content), + "file_hash_sha256": hashlib.sha256(content).hexdigest(), "storage_backend": "LOCAL_YEAR_WISE", + "local_relative_path": f"UAT-A/UAT-CL-A/{ACTIVE_FY}/{filename}", "storage_status": "stored", + "uploaded_by_user_id": staff_id, + }, f"document version {version}") + doc_deleted = upsert(documents, {"tenant_id": tenant_a_id, "document_code": "UAT-DOC-DELETED-001"}, { + **common_scope, "branch_id": branch_a_id, "client_id": client_a_id, "engagement_id": sub_a_id, + "financial_year": ACTIVE_FY, "document_type": "SECURITY", "title": "UAT Deleted Document", + "status": "deleted", "is_deleted": True, "deleted_at": utcnow(), "deleted_by_user_id": firm_admin_id, + }, "deleted document") + ids["DELETED_DOCUMENT_ID"] = val(doc_deleted, "id") + + storage_job = upsert(storage_jobs, {"tenant_id": tenant_a_id, "job_code": "UAT-STORAGE-JOB-SEC-001"}, { + **common_scope, "storage_node_id": val(node, "id"), "document_id": val(doc_sec, "id"), + "job_type": "store_version", "status": "pending", "priority": 1, + "staging_relative_path": "staging/uat-sec-v2.txt", "target_relative_path": f"UAT-A/UAT-CL-A/{ACTIVE_FY}/uat-sec-v2.txt", + "file_size_bytes": 21, "expected_hash_sha256": hashlib.sha256(b"UAT secure version 2\n").hexdigest(), + "attempts": 0, + }, "storage job security") + ids["STORAGE_JOB_SECURITY_ID"] = val(storage_job, "id") + + # 8. HR seed + attendance = find_table("employee_attendance", "attendance", "employee_attendances") + upsert(attendance, {"tenant_id": tenant_a_id, "employee_id": val(first(employees, user_id=staff_id), "id"), "attendance_date": today()}, { + **common_scope, "user_id": staff_id, "status": "present", "check_in_time": utcnow(), "source": "seed", + "remarks": "Seed attendance for duplicate/manager tests", + }, "attendance") + leave_balances = find_table("employee_leave_balances", "leave_balances") + leave_balance = upsert(leave_balances, {"tenant_id": tenant_a_id, "user_id": staff_id, "leave_type": "casual", "financial_year": ACTIVE_FY}, { + **common_scope, "employee_id": val(first(employees, user_id=staff_id), "id"), "opening_balance": 12, + "availed": 2, "balance": 10, "is_active": True, + }, "leave balance") + leave_requests = find_table("employee_leave_requests", "leave_requests", "leaves") + leave_req = upsert(leave_requests, {"tenant_id": tenant_a_id, "leave_code": "UAT-LEAVE-PENDING-001"}, { + **common_scope, "user_id": staff_id, "employee_id": val(first(employees, user_id=staff_id), "id"), + "leave_type": "casual", "from_date": today() + timedelta(days=5), "to_date": today() + timedelta(days=6), + "days": 2, "status": "pending", "reason": "UAT seeded pending leave", "approver_user_id": manager_id, + "balance_id": val(leave_balance, "id"), + }, "leave request pending") + ids["PENDING_LEAVE_ID"] = val(leave_req, "id") + payroll_runs = find_table("payroll_runs", "employee_payroll_runs", "payroll") + payroll = upsert(payroll_runs, {"tenant_id": tenant_a_id, "run_code": "UAT-PAYROLL-001"}, { + **common_scope, "financial_year": ACTIVE_FY, "period_month": today().month, "period_year": today().year, + "status": "generated", "gross_amount": 50000, "deduction_amount": 5000, "net_amount": 45000, + }, "payroll run") + payslips = find_table("payslips", "employee_payslips") + upsert(payslips, {"tenant_id": tenant_a_id, "payslip_code": "UAT-PAYSLIP-001"}, { + **common_scope, "payroll_run_id": val(payroll, "id"), "employee_id": val(first(employees, user_id=staff_id), "id"), + "user_id": staff_id, "gross_amount": 50000, "deduction_amount": 5000, "net_amount": 45000, + "status": "generated", "file_name": "uat-payslip-001.pdf", + }, "payslip") + + # 9. Firm billing business seed + firm_invoices = find_table("billing_invoices", "firm_invoices", "client_invoices", "invoices") + draft_inv = upsert(firm_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-BILL-DRAFT-001"}, { + **common_scope, "client_id": client_a_id, "invoice_date": today(), "due_date": today() + timedelta(days=15), + "status": "draft", "subtotal": 10000, "tax_amount": 1800, "total_amount": 11800, "balance_amount": 11800, + "financial_year": ACTIVE_FY, + }, "billing draft invoice") + issued_inv = upsert(firm_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-BILL-ISSUED-001"}, { + **common_scope, "client_id": client_a_id, "invoice_date": today(), "due_date": today() + timedelta(days=15), + "status": "issued", "subtotal": 20000, "tax_amount": 3600, "total_amount": 23600, "balance_amount": 18600, + "financial_year": ACTIVE_FY, + }, "billing issued invoice") + cancel_inv = upsert(firm_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-BILL-CANCELLED-001"}, { + **common_scope, "client_id": client_a_id, "invoice_date": today(), "status": "cancelled", + "subtotal": 1000, "tax_amount": 180, "total_amount": 1180, "balance_amount": 0, + "financial_year": ACTIVE_FY, + }, "billing cancelled invoice") + firm_payments = find_table("billing_payments", "client_payments", "invoice_payments", "payments") + upsert(firm_payments, {"tenant_id": tenant_a_id, "payment_ref": "UAT-BILL-PAY-001"}, { + **common_scope, "invoice_id": val(issued_inv, "id"), "client_id": client_a_id, + "payment_date": today(), "amount": 5000, "payment_mode": "upi", "status": "posted", + }, "billing partial payment") + ids["BILLING_DRAFT_INVOICE_ID"] = val(draft_inv, "id") + ids["BILLING_ISSUED_INVOICE_ID"] = val(issued_inv, "id") + ids["BILLING_CANCELLED_INVOICE_ID"] = val(cancel_inv, "id") + + # 10. Notice case business seed + if notice_cases is not None: + nc = upsert(notice_cases, {"tenant_id": tenant_a_id, "case_code": "UAT-NC-DEEP-001"}, { + **common_scope, "branch_id": branch_a_id, "client_id": client_a_id, "reference_no": "UAT-NOTICE-DEEP-001", + "department": "GST", "case_type": "Notice", "title": "UAT Deep GST Notice", "status": "Hearing Scheduled", + "notice_date": today() - timedelta(days=10), "hearing_date": today() + timedelta(days=10), + "order_date": None, "due_date": today() + timedelta(days=20), "assigned_to_user_id": staff_id, + "partner_user_id": partner_id, "internal_notes": "Internal note must not be client-visible", + "client_visible_notes": "Client visible notice update", + "issue_summary": "Seeded v2.5.1 notice case for business-rule testing", + }, "deep notice case") + ids["NOTICE_CASE_DEEP_ID"] = val(nc, "id") + nc_events = find_table("notice_case_events", "case_events", "notice_case_activities") + upsert(nc_events, {"tenant_id": tenant_a_id, "event_code": "UAT-NC-EVT-001"}, { + **common_scope, "case_id": val(nc, "id"), "event_type": "hearing_scheduled", "event_date": today(), + "notes": "Seeded hearing event", "is_client_visible": True, "created_by_user_id": staff_id, + }, "notice event") + + db.commit() + + print("\n[v2.5.1 seed] Completed seed extension.") + print("[v2.5.1 seed] Created/updated data for email, marketplace, platform billing, RBAC, audit, FY, documents, HR, billing and notice-case tests.") + print("\nUseful IDs for Playwright .env if needed:\n") + for k, v in ids.items(): + if v is not None: + print(f"{k}={v}") + print("\nSkipped/missing-module notes:") + for line in report: + print(f"- {line}") + print("\nIf many rows were skipped, that means the corresponding ERP module/table is not present yet; tests for those modules should then fail/skip as ERP gaps, not seed failures.") + + db.close() + + +if __name__ == "__main__": + try: + main() + except Exception: + raise