A R R R Associates
458 lines
18 KiB
Python
458 lines
18 KiB
Python
"""
|
|
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()
|