Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add skipped test seed data script

Browse Source
This commit is contained in:
A R R R Associates
2026-06-27 13:03:23 +05:30
parent d347485976
commit f5174575c5
1 changed files with 457 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/seed_uat_skipped_tests_FIX6.py
+457
View File
@@ -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()