""" Audit Firm ERP -- UAT seed patch for skipped Playwright tests. Run inside ERP container AFTER core seed and v2.5.1 seed: cd /app PYTHONPATH=/app python scripts/seed_uat_skipped_tests_FIX6.py Purpose: Adds the missing seed records that caused skipped tests in: - employees.spec.js - consultants-documents.spec.js - noticecases-services-work.spec.js Safe/idempotent for UAT-coded rows. Do NOT run against live production data. """ from __future__ import annotations import hashlib 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") ASSESSMENT_YEAR = os.getenv("ASSESSMENT_YEAR", "2026-27") def fail(msg: str) -> None: print(f"\n[skipped-seed FIX6] 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(f"Could not import ERP DB modules. Run 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] = [] ids: dict[str, object] = {} def norm(name: str) -> str: return name.lower().replace("_", "").replace("-", "") def find_table(*names: str): for name in names: if name in meta.tables: return meta.tables[name] normalized = {norm(t): t for t in table_names} for name in names: hit = normalized.get(norm(name)) if hit: return meta.tables[hit] 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 where_clause(table, lookup: dict): data = valid(table, lookup) if not data: return None return and_(*[table.c[k] == v for k, v in data.items()]) def one(table, lookup: dict): if table is None: return None clause = where_clause(table, lookup) if clause is None: return None return db.execute(select(table).where(clause)).mappings().first() def first(table, **lookup): return one(table, lookup) def first_any(table, lookups: list[dict]): for lookup in lookups: row = one(table, lookup) if row: return row return None def first_row(table): if table is None: return None return db.execute(select(table).limit(1)).mappings().first() def val(row, key: str, default=None): return row[key] if row is not None and key in row else default def default_for_required(table, col_name: str): name = col_name.lower() if name.endswith("_at_utc") or name.endswith("_at"): return utcnow() if name.endswith("_date"): return today() if name in {"status"}: return "active" if name in {"is_active"}: return True if name in {"is_deleted"}: return False try: pytype = table.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 today() if pytype is datetime: return utcnow() return f"UAT {col_name.replace('_', ' ').title()}" def add_required_defaults(table, data: dict) -> dict: 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 out[col.name] = default_for_required(table, col.name) return valid(table, out) def upsert(table, lookup: dict, defaults: dict, label: str): if table is None: notes.append(f"SKIP {label}: table missing") return None lookup_v = valid(table, lookup) defaults_v = valid(table, defaults) if not lookup_v: notes.append(f"SKIP {label}: lookup columns missing in {table.name}") return None row = one(table, lookup_v) if row: if defaults_v: db.execute(update(table).where(where_clause(table, lookup_v)).values(**defaults_v)) db.flush() return one(table, lookup_v) data = add_required_defaults(table, {**lookup_v, **defaults_v}) try: 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) except Exception as exc: notes.append(f"SKIP {label}: insert failed in {table.name}: {exc.__class__.__name__}: {str(exc).splitlines()[0]}") db.rollback() return one(table, lookup_v) # Core tables tenants = find_table("tenants") branches = find_table("branches") users = find_table("users") clients = find_table("clients") employees = find_table("employees") services = find_table("service_catalogues", "services") notice_cases = find_table("notice_cases") tasks = find_table("client_service_task_instances") subscriptions = find_table("client_service_subscriptions") tenant_a = first_any(tenants, [{"code": "UAT-A"}, {"tenant_code": "UAT-A"}, {"name": "UAT Tenant A"}]) branch_a = first_any(branches, [{"code": "UAT-BA"}, {"branch_code": "UAT-BA"}, {"name": "UAT Branch A"}]) firm_admin = first(users, email=f"uat.firmadmin@{UAT_DOMAIN}") or first(users, email=f"uat.admin@{UAT_DOMAIN}") staff = first(users, email=f"uat.staff@{UAT_DOMAIN}") or firm_admin manager = first(users, email=f"uat.manager@{UAT_DOMAIN}") or firm_admin partner = first(users, email=f"uat.partner@{UAT_DOMAIN}") or firm_admin consultant_user = first(users, email=f"uat.consultant@{UAT_DOMAIN}") or firm_admin client_a = first_any(clients, [{"client_code": "UAT-CL-A"}, {"code": "UAT-CL-A"}, {"client_name": "UAT Client A Pvt Ltd"}]) or first_row(clients) service_a = first_any(services, [{"service_code": "GST-GSTR3B-M"}, {"code": "GST-GSTR3B-M"}]) or first_row(services) notice_case = first_any(notice_cases, [{"case_code": "UAT-NC-DEEP-001"}, {"reference_no": "UAT-NOTICE-DEEP-001"}]) or first_row(notice_cases) task_a = first_row(tasks) sub_a = first_row(subscriptions) employee_a = first(employees, user_id=val(staff, "id")) or first_row(employees) if not tenant_a or not branch_a or not firm_admin: fail("Core seed is missing. Run seed_uat_data.py first.") tenant_id = val(tenant_a, "id") branch_id = val(branch_a, "id") admin_id = val(firm_admin, "id") staff_id = val(staff, "id", admin_id) manager_id = val(manager, "id", admin_id) partner_id = val(partner, "id", admin_id) consultant_user_id = val(consultant_user, "id", admin_id) client_id = val(client_a, "id") service_id = val(service_a, "id") employee_id = val(employee_a, "id") notice_case_id = val(notice_case, "id") scope = { "tenant_id": tenant_id, "branch_id": branch_id, "created_by_user_id": admin_id, "updated_by_user_id": admin_id, "created_by_id": admin_id, "updated_by_id": admin_id, "created_at_utc": utcnow(), "updated_at_utc": utcnow(), "is_active": True, } # 1) Employee registration request employee_registration_requests = find_table("employee_registration_requests") reg = upsert(employee_registration_requests, {"tenant_id": tenant_id, "requested_employee_code": "UAT-REG-STAFF-001"}, { **scope, "user_id": staff_id, "full_name": "UAT Staff Registration Request", "email": f"uat.staff@{UAT_DOMAIN}", "mobile": "9000000001", "department": "Audit", "designation": "Audit Staff", "date_of_joining": today(), "remarks": "Seeded pending registration request for Playwright skipped tests", "status": "pending", }, "employee registration request") ids["REG_REQUEST_ID"] = val(reg, "id") # 2) Employee document + document type emp_doc_types = find_table("employee_document_types") doc_type = upsert(emp_doc_types, {"tenant_id": tenant_id, "code": "UAT-KYC"}, { **scope, "name": "UAT KYC Document", "description": "Seed document type for employee tests", "is_mandatory": False, "status": "active", }, "employee document type") emp_docs = find_table("employee_documents") emp_doc = upsert(emp_docs, {"tenant_id": tenant_id, "employee_id": employee_id, "title": "UAT Employee KYC Document"}, { **scope, "employee_id": employee_id, "document_type_id": val(doc_type, "id"), "title": "UAT Employee KYC Document", "document_no": "UAT-EMP-DOC-001", "issue_date": today(), "original_filename": "uat-employee-kyc.txt", "stored_filename": "uat-employee-kyc.txt", "storage_path": "employee_documents/uat-employee-kyc.txt", "content_type": "text/plain", "file_size_bytes": 27, "status": "uploaded", "visibility": "employee_and_hr", "remarks": "Seeded employee document for Playwright skipped tests", "uploaded_by_user_id": admin_id, "updated_by_user_id": admin_id, }, "employee document") ids["EMP_DOCUMENT_ID"] = val(emp_doc, "id") # 3) Consultant profile, managed conversion record, service request consultant_profiles = find_table("consultant_profiles") consultant = upsert(consultant_profiles, {"tenant_id": tenant_id, "email": f"uat.consultant@{UAT_DOMAIN}"}, { **scope, "branch_id": branch_id, "user_id": consultant_user_id, "consultant_type": "external_consultant", "firm_name": "UAT Consultant Firm", "contact_person": "UAT Consultant", "email": f"uat.consultant@{UAT_DOMAIN}", "mobile": "9000000002", "specialisation": "GST and Income Tax", "pan": "ABCDE1234F", "address": "UAT Consultant Address", "status": "active", "onboarding_status": "approved", "is_platform_partner": False, "is_franchise_partner": False, "is_saas_customer": False, "is_active": True, }, "consultant profile") ids["CONSULTANT_A_ID"] = val(consultant, "id") consultant_managed_clients = find_table("consultant_managed_clients") managed_client = upsert(consultant_managed_clients, {"tenant_id": tenant_id, "consultant_id": val(consultant, "id"), "client_code": "UAT-CMC-CONV-001"}, { **scope, "consultant_id": val(consultant, "id"), "client_code": "UAT-CMC-CONV-001", "client_name": "UAT Consultant Managed Client", "trade_name": "UAT Managed Client", "client_type": "Private Limited Company", "pan": "ABCDE9999F", "gstin": "33ABCDE9999F1Z5", "contact_person_name": "UAT Managed Contact", "mobile": "9000000003", "email": "uat.managed.client@example.com", "address_line_1": "UAT Managed Client Address", "city": "Chennai", "state": "Tamil Nadu", "pincode": "600001", "country": "India", "service_interest": "GST compliance", "relationship_stage": "conversion_requested", "status": "active", "is_active": True, "conversion_status": "requested", "conversion_requested_at_utc": utcnow(), "conversion_requested_by_user_id": consultant_user_id, "conversion_notes": "Seeded conversion request for Playwright tests", }, "consultant managed conversion request") # Playwright route /consultants/conversion-requests/{id} expects managed client id. ids["CONVERSION_REQUEST_ID"] = val(managed_client, "id") service_requests = find_table("consultant_service_requests") service_request = upsert(service_requests, {"tenant_id": tenant_id, "request_no": "UAT-CSR-001"}, { **scope, "branch_id": branch_id, "consultant_id": val(consultant, "id"), "managed_client_id": val(managed_client, "id"), "firm_client_id": client_id, "service_catalogue_id": service_id, "request_no": "UAT-CSR-001", "request_type": "service_request", "status": "submitted", "priority": "normal", "requested_service_name": "GST Monthly Filing", "requested_due_date": today() + timedelta(days=15), "subject": "UAT Consultant Service Request", "description": "Seeded consultant service request for Playwright skipped tests", "consultant_notes": "UAT seed", "created_by_user_id": consultant_user_id, "updated_by_user_id": admin_id, }, "consultant service request") ids["SERVICE_REQUEST_ID"] = val(service_request, "id") # 4) Permanent client document and version permanent_docs = find_table("permanent_client_documents") permanent_doc = upsert(permanent_docs, {"tenant_id": tenant_id, "client_id": client_id, "document_code": "UAT-PERM-DOC-001"}, { **scope, "client_id": client_id, "document_code": "UAT-PERM-DOC-001", "category": "KYC", "title": "UAT Permanent Client Document", "description": "Seeded permanent document for Playwright skipped tests", "current_version_no": 1, "status": "active", "is_deleted": False, }, "permanent client document") ids["PERM_DOCUMENT_ID"] = val(permanent_doc, "id") perm_versions = find_table("permanent_client_document_versions") content = b"UAT permanent document content\n" perm_version = upsert(perm_versions, {"document_id": val(permanent_doc, "id"), "version_no": 1}, { "document_id": val(permanent_doc, "id"), "tenant_id": tenant_id, "branch_id": branch_id, "client_id": client_id, "version_no": 1, "original_filename": "uat-permanent-document.txt", "stored_filename": "uat-permanent-document.txt", "content_type": "text/plain", "file_size_bytes": len(content), "file_hash_sha256": hashlib.sha256(content).hexdigest(), "storage_backend": "LOCAL_PERMANENT", "local_relative_path": "UAT-A/UAT-CL-A/permanent/uat-permanent-document.txt", "storage_status": "stored", "remarks": "Seeded permanent document version for Playwright tests", "uploaded_by_user_id": admin_id, "uploaded_at_utc": utcnow(), }, "permanent client document version") ids["PERM_DOCUMENT_VERSION_ID"] = val(perm_version, "id") # 5) Notice case document notice_docs = find_table("notice_case_documents") notice_doc = upsert(notice_docs, {"tenant_id": tenant_id, "case_id": notice_case_id, "title": "UAT Notice Case Document"}, { "tenant_id": tenant_id, "branch_id": branch_id, "case_id": notice_case_id, "document_type": "GENERAL", "title": "UAT Notice Case Document", "description": "Seeded notice case document for Playwright skipped tests", "original_filename": "uat-notice-document.txt", "stored_filename": "uat-notice-document.txt", "content_type": "text/plain", "file_size_bytes": 26, "local_relative_path": "notice_cases/uat-notice-document.txt", "version_no": 1, "status": "active", "is_deleted": False, "uploaded_by_user_id": admin_id, "uploaded_at_utc": utcnow(), }, "notice case document") ids["CASE_DOCUMENT_A_ID"] = val(notice_doc, "id") db.commit() # Also print existing IDs required by skipped modules. def scalar(sql: str): from sqlalchemy import text try: return db.execute(text(sql)).scalar() except Exception: return None env_ids = { "EMPLOYEE_A_ID": scalar("SELECT id FROM employees ORDER BY id DESC LIMIT 1"), "ATTENDANCE_ID": scalar("SELECT id FROM employee_attendance ORDER BY id DESC LIMIT 1"), "LEAVE_REQUEST_ID": scalar("SELECT id FROM employee_leave_requests ORDER BY id DESC LIMIT 1"), "REG_REQUEST_ID": ids.get("REG_REQUEST_ID"), "EMP_DOCUMENT_ID": ids.get("EMP_DOCUMENT_ID"), "PAYROLL_RUN_ID": scalar("SELECT id FROM employee_payroll_runs ORDER BY id DESC LIMIT 1"), "CONSULTANT_A_ID": ids.get("CONSULTANT_A_ID"), "CONVERSION_REQUEST_ID": ids.get("CONVERSION_REQUEST_ID"), "SERVICE_REQUEST_ID": ids.get("SERVICE_REQUEST_ID"), "PERM_DOCUMENT_ID": ids.get("PERM_DOCUMENT_ID"), "STORAGE_NODE_ID": scalar("SELECT id FROM branch_storage_nodes ORDER BY id DESC LIMIT 1"), "CASE_DOCUMENT_A_ID": ids.get("CASE_DOCUMENT_A_ID"), "SUBSCRIPTION_A_ID": scalar("SELECT id FROM client_service_subscriptions ORDER BY id DESC LIMIT 1"), "TASK_A_ID": scalar("SELECT id FROM client_service_task_instances ORDER BY id DESC LIMIT 1"), "INVOICE_A_ID": scalar("SELECT id FROM billing_invoices ORDER BY id DESC LIMIT 1"), "PAYMENT_A_ID": scalar("SELECT id FROM billing_payments ORDER BY id DESC LIMIT 1"), "PARTNER_TASK_A_ID": scalar("SELECT id FROM client_service_task_instances WHERE review_partner_user_id IS NOT NULL ORDER BY id DESC LIMIT 1"), "LOCKED_FY": "2024-25", "EXPECT_STRICT_CSP": "false", } print("\n[skipped-seed FIX6] Completed missing skipped-test seed patch.") print("\nCopy these into /tests/.env:\n") for k, v in env_ids.items(): print(f"{k}={v or ''}") print("\nNotes:") if notes: for n in notes: print(f"- {n}") else: print("- No skipped seed patch errors reported.") db.close() if __name__ == "__main__": main()