diff --git a/scripts/seed_uat_skipped_tests_FIX6.py b/scripts/seed_uat_skipped_tests_FIX6.py new file mode 100644 index 0000000..2820f93 --- /dev/null +++ b/scripts/seed_uat_skipped_tests_FIX6.py @@ -0,0 +1,457 @@ +""" +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()