""" Audit Firm ERP -- UAT seed data for Playwright v2.4.1 IMAP. Purpose: - Replace old @tenant-a.test / @tenant-b.test UAT users with real @vavalam.com test emails. - Create all Playwright role users. - Create tenants, branches, clients, employees/staff, service catalogue, task templates, subscriptions/engagements, task instances, notice case, and a local storage node. Run inside ERP container from project root after migrations: cd /app PYTHONPATH=/app UAT_SEED_PASSWORD='Password@123' python scripts/seed_uat_data.py Safe/idempotent for UAT: re-running updates/reuses existing UAT records. Do NOT run against live production data. """ from __future__ import annotations import hashlib import os import sys from datetime import date, datetime, timedelta, timezone PASSWORD = os.getenv("UAT_SEED_PASSWORD", "Password@123") 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") ROLE_NAME = { "system_admin": "System Admin", "firm_admin": "Firm Admin", "partner": "Partner", "manager": "Branch Manager", "staff": "Staff", "client": "Client", "consultant": "Consultant", } LEGACY_EMAIL_MAP = { "uat.firmadmin@tenant-a.test": "uat.firmadmin@vavalam.com", "uat.partner@tenant-a.test": "uat.partner@vavalam.com", "uat.manager@tenant-a.test": "uat.manager@vavalam.com", "uat.staff@tenant-a.test": "uat.staff@vavalam.com", "uat.client@tenant-a.test": "uat.client@vavalam.com", "uat.consultant@tenant-a.test": "uat.consultant@vavalam.com", "uat.firmadmin@tenant-b.test": "uat.firmadminb@vavalam.com", } def fail(msg: str) -> None: print(f"\n[seed] ERROR: {msg}", file=sys.stderr) sys.exit(1) def main() -> None: try: from app.core.db.common import CommonSessionLocal from app.core.security.passwords import hash_password from app.modules.core.iam.models import User from app.modules.core.tenancy.models import Tenant, Branch, FinancialYear from app.modules.core.rbac.models import Role, UserRole from app.modules.clients.models import Client from app.modules.employees.models import Employee from app.modules.services.models import ( ServiceCategory, ServiceCatalogue, ServiceDefaultTaskTemplate, FirmServiceSelection, FirmServiceTaskTemplate, FirmTaskDocumentRequirement, ServiceDueDateRule, ClientServiceSubscription, ClientServiceTaskInstance, ) from app.modules.documents.models import ( BranchStorageNode, EngagementDocument, EngagementDocumentVersion, DocumentStorageJob, ) except Exception as exc: fail( "Could not import ERP modules. Run this from ERP project root inside the app container. " f"Underlying import error: {exc!r}" ) try: from app.modules.notice_cases.models import NoticeCase except Exception: NoticeCase = None db = CommonSessionLocal() ids: dict[str, object] = {} def valid_data(Model, data: dict) -> dict: valid = {c.name for c in Model.__table__.columns} return {k: v for k, v in data.items() if k in valid} def get_or_create(Model, lookup: dict, defaults: dict | None = None): obj = db.query(Model).filter_by(**lookup).first() if obj: if defaults: for k, v in valid_data(Model, defaults).items(): setattr(obj, k, v) db.flush() return obj, False data = {**lookup, **(defaults or {})} obj = Model(**valid_data(Model, data)) db.add(obj) db.flush() return obj, True def role_by_name(name: str): return db.query(Role).filter(Role.name == name).first() def assign_role(user, role_label: str) -> None: role_name = ROLE_NAME[role_label] role = role_by_name(role_name) if not role: print(f"[seed] WARNING: role '{role_name}' not found. Run/restart ERP startup seed first.") return exists = db.query(UserRole).filter_by(user_id=user.id, role_id=role.id).first() if not exists: db.add(UserRole(user_id=user.id, role_id=role.id)) db.flush() def deactivate_user(email: str) -> None: user = db.query(User).filter(User.email == email).first() if user: user.allow_login = False user.is_active = False if hasattr(user, "deleted_at"): user.deleted_at = datetime.now(timezone.utc) db.flush() def ensure_user(email: str, full_name: str, role_labels: list[str], tenant, branch, *, locked=False, allow_login=True): # If old fake email exists and mapped to this real email, update old record in-place unless real already exists. for old_email, new_email in LEGACY_EMAIL_MAP.items(): if new_email == email: old_user = db.query(User).filter(User.email == old_email).first() real_user = db.query(User).filter(User.email == email).first() if old_user and not real_user: old_user.email = email old_user.full_name = full_name old_user.tenant_id = tenant.id old_user.branch_id = branch.id old_user.password_hash = hash_password(PASSWORD) old_user.is_active = True old_user.allow_login = allow_login old_user.is_locked = locked old_user.must_change_password = False db.flush() for role_label in role_labels: assign_role(old_user, role_label) return old_user if old_user and real_user and old_user.id != real_user.id: deactivate_user(old_email) user, _ = get_or_create( User, {"email": email}, { "full_name": full_name, "password_hash": hash_password(PASSWORD), "tenant_id": tenant.id, "branch_id": branch.id, "is_active": True, "allow_login": allow_login, "is_locked": locked, "must_change_password": False, "mobile": "9000000000", }, ) user.password_hash = hash_password(PASSWORD) user.full_name = full_name user.tenant_id = tenant.id user.branch_id = branch.id user.is_active = True user.allow_login = allow_login user.is_locked = locked user.must_change_password = False for role_label in role_labels: assign_role(user, role_label) db.flush() return user def ensure_employee(code: str, user, tenant, branch, designation: str, department="Audit"): emp, _ = get_or_create( Employee, {"tenant_id": tenant.id, "employee_code": code}, { "branch_id": branch.id, "user_id": user.id, "full_name": user.full_name, "email": user.email, "mobile": getattr(user, "mobile", None) or "9000000000", "date_of_joining": date(2025, 4, 1), "employment_type": "full_time", "status": "active", "is_active": True, "department": department, "designation": designation, "pan": f"UAT{code[-3:].zfill(3)}P", }, ) return emp def ensure_fy(tenant, year_code: str, ay: str, start: date, end: date, current: bool): fy, _ = get_or_create( FinancialYear, {"tenant_id": tenant.id, "year_code": year_code}, { "assessment_year": ay, "start_date": start, "end_date": end, "is_current": current, "is_locked": False, }, ) return fy def ensure_service_category(code: str, name: str, sort_order: int): return get_or_create( ServiceCategory, {"code": code}, {"name": name, "sort_order": sort_order, "is_active": True}, )[0] def ensure_catalogue(code: str, name: str, category_obj, recurrence: str, engagement_type="non_assurance", sort_order=100): svc, _ = get_or_create( ServiceCatalogue, {"service_code": code}, { "service_name": name, "category": category_obj.name, "category_id": category_obj.id, "recurrence_type": recurrence, "engagement_type": engagement_type, "sort_order": sort_order, "description": f"UAT seed service: {name}", "applicable_individual": True, "applicable_proprietorship": True, "applicable_partnership": True, "applicable_llp": True, "applicable_company": True, "is_active": True, "is_client_requestable": True, "is_consultant_requestable": False, }, ) return svc def ensure_default_task(svc, seq: int, name: str, role: str, review=False): return get_or_create( ServiceDefaultTaskTemplate, {"service_catalogue_id": svc.id, "sequence_no": seq}, { "task_name": name, "description": f"UAT default task: {name}", "default_role_name": role, "is_mandatory": True, "requires_review": review, "is_active": True, }, )[0] def enable_firm_service(tenant, branch, svc, user): return get_or_create( FirmServiceSelection, {"tenant_id": tenant.id, "service_catalogue_id": svc.id}, { "is_enabled": True, "default_branch_id": branch.id, "activated_by_user_id": user.id, "updated_by_user_id": user.id, }, )[0] def ensure_firm_task(tenant, svc, seq: int, name: str, role: str, user, review=False): task, _ = get_or_create( FirmServiceTaskTemplate, {"tenant_id": tenant.id, "service_catalogue_id": svc.id, "sequence_no": seq}, { "task_name": name, "description": f"UAT firm task: {name}", "default_role_name": role, "is_mandatory": True, "requires_review": review, "is_active": True, "created_by_user_id": user.id, "updated_by_user_id": user.id, }, ) return task def ensure_doc_requirement(tenant, svc, task, name: str, dtype="GENERAL"): return get_or_create( FirmTaskDocumentRequirement, {"tenant_id": tenant.id, "firm_task_template_id": task.id, "document_name": name}, { "service_catalogue_id": svc.id, "document_type": dtype, "is_mandatory": True, "allowed_file_types": "pdf,xlsx,docx,jpg,png,txt", "instructions": "UAT seeded document requirement", "sort_order": task.sequence_no, "is_active": True, }, )[0] def ensure_due_rule(svc, name: str, period_type: str, day: int | None, month: int | None, offset=0): return get_or_create( ServiceDueDateRule, {"service_catalogue_id": svc.id, "rule_name": name}, { "period_type": period_type, "due_year_basis": "financial_year_start", "due_day": day, "due_month": month, "due_month_offset": offset, "sort_order": 100, "is_active": True, "remarks": "UAT seed due date rule", }, )[0] def ensure_subscription(tenant, branch, client, svc, firm_selection, partner, manager, staff, due_rule=None): sub, _ = get_or_create( ClientServiceSubscription, {"tenant_id": tenant.id, "client_id": client.id, "service_catalogue_id": svc.id, "financial_year": ACTIVE_FY}, { "branch_id": branch.id, "firm_service_selection_id": firm_selection.id, "assigned_partner_user_id": partner.id, "assigned_manager_user_id": manager.id, "assigned_staff_user_id": staff.id, "review_partner_user_id": partner.id, "assessment_year": ASSESSMENT_YEAR, "engagement_type": svc.engagement_type, "due_date_rule_id": due_rule.id if due_rule else None, "original_due_date": date.today() + timedelta(days=30), "current_due_date": date.today() + timedelta(days=30), "due_date_source": "seed", "start_date": date.today(), "end_date": date.today() + timedelta(days=365), "status": "active", "remarks": "UAT seeded engagement/subscription", "is_active": True, "is_locked": False, "created_by_user_id": partner.id, "updated_by_user_id": partner.id, }, ) return sub def ensure_task_instance(tenant, branch, client, sub, svc, firm_task, assigned_user, days=7): task, _ = get_or_create( ClientServiceTaskInstance, {"subscription_id": sub.id, "firm_task_template_id": firm_task.id, "financial_year": ACTIVE_FY}, { "tenant_id": tenant.id, "branch_id": branch.id, "client_id": client.id, "service_catalogue_id": svc.id, "assessment_year": ASSESSMENT_YEAR, "task_name": firm_task.task_name, "description": firm_task.description, "sequence_no": firm_task.sequence_no, "default_role_name": firm_task.default_role_name, "assigned_to_user_id": assigned_user.id, "internal_target_date": date.today() + timedelta(days=days), "status": "pending", "priority": "normal", "is_active": True, "is_locked": False, "created_by_user_id": assigned_user.id, "updated_by_user_id": assigned_user.id, }, ) return task try: # Tenants and branches tenant_a, _ = get_or_create(Tenant, {"code": "UAT-A"}, {"name": "UAT Tenant A", "is_active": True, "firm_type": "partnership", "contact_email": "support@vavalam.com"}) tenant_b, _ = get_or_create(Tenant, {"code": "UAT-B"}, {"name": "UAT Tenant B", "is_active": True, "firm_type": "partnership", "contact_email": "support@vavalam.com"}) branch_a, _ = get_or_create(Branch, {"tenant_id": tenant_a.id, "code": "UAT-BA"}, {"name": "UAT Branch A", "is_active": True, "is_head_office": True, "timezone": "Asia/Kolkata", "local_storage_path": "D:/UAT/BranchA"}) branch_b, _ = get_or_create(Branch, {"tenant_id": tenant_b.id, "code": "UAT-BB"}, {"name": "UAT Branch B", "is_active": True, "is_head_office": True, "timezone": "Asia/Kolkata", "local_storage_path": "D:/UAT/BranchB"}) ids["TENANT_A_ID"] = tenant_a.id ids["TENANT_B_ID"] = tenant_b.id ids["BRANCH_A_ID"] = branch_a.id ids["BRANCH_B_ID"] = branch_b.id fy_a = ensure_fy(tenant_a, ACTIVE_FY, ASSESSMENT_YEAR, date(2025, 4, 1), date(2026, 3, 31), True) ensure_fy(tenant_a, PREVIOUS_FY, "2025-26", date(2024, 4, 1), date(2025, 3, 31), False) fy_b = ensure_fy(tenant_b, ACTIVE_FY, ASSESSMENT_YEAR, date(2025, 4, 1), date(2026, 3, 31), True) ensure_fy(tenant_b, PREVIOUS_FY, "2025-26", date(2024, 4, 1), date(2025, 3, 31), False) ids["ACTIVE_FY_ID"] = fy_a.id # Users: real IMAP-capable emails system_admin = ensure_user("uat.admin@vavalam.com", "UAT System Admin", ["system_admin"], tenant_a, branch_a) firm_admin = ensure_user("uat.firmadmin@vavalam.com", "UAT Firm Admin", ["firm_admin"], tenant_a, branch_a) partner = ensure_user("uat.partner@vavalam.com", "UAT Partner", ["partner"], tenant_a, branch_a) manager = ensure_user("uat.manager@vavalam.com", "UAT Manager", ["manager"], tenant_a, branch_a) staff = ensure_user("uat.staff@vavalam.com", "UAT Staff", ["staff"], tenant_a, branch_a) staff2 = ensure_user("uat.staff2@vavalam.com", "UAT Staff 2", ["staff"], tenant_a, branch_a) branch2staff = ensure_user("uat.branch2staff@vavalam.com", "UAT Branch 2 Staff", ["staff"], tenant_b, branch_b) client_user = ensure_user("uat.client@vavalam.com", "UAT Client User", ["client"], tenant_a, branch_a) client2_user = ensure_user("uat.client2@vavalam.com", "UAT Client 2 User", ["client"], tenant_b, branch_b) consultant = ensure_user("uat.consultant@vavalam.com", "UAT Consultant", ["consultant"], tenant_a, branch_a) invite_user = ensure_user("uat.invite@vavalam.com", "UAT Invite User", ["staff"], tenant_a, branch_a, allow_login=True) reset_user = ensure_user("uat.reset@vavalam.com", "UAT Reset User", ["staff"], tenant_a, branch_a, allow_login=True) locked_user = ensure_user("uat.lockeduser@vavalam.com", "UAT Locked User", ["staff"], tenant_a, branch_a, locked=True, allow_login=True) # Optional but recommended: create one true Tenant B admin for cross-tenant tests. # Please create this mailbox if OTP is required for Tenant B Admin login. tenant_b_admin = ensure_user("uat.firmadminb@vavalam.com", "UAT Firm Admin B", ["firm_admin"], tenant_b, branch_b) for old in LEGACY_EMAIL_MAP: if not LEGACY_EMAIL_MAP[old].endswith("@vavalam.com"): deactivate_user(old) # Employees/staff master records ensure_employee("UAT-MGR-A", manager, tenant_a, branch_a, "Branch Manager") ensure_employee("UAT-STF-A", staff, tenant_a, branch_a, "Audit Staff") ensure_employee("UAT-STF-A2", staff2, tenant_a, branch_a, "Audit Staff") ensure_employee("UAT-STF-B", branch2staff, tenant_b, branch_b, "Branch Staff") ensure_employee("UAT-LOCK-A", locked_user, tenant_a, branch_a, "Locked Test User") # Clients client_a, _ = get_or_create( Client, {"tenant_id": tenant_a.id, "client_code": "UAT-CL-A"}, { "branch_id": branch_a.id, "partner_id": partner.id, "default_review_partner_user_id": partner.id, "portal_user_id": client_user.id, "engagement_mode": "hybrid", "client_name": "UAT Client A Pvt Ltd", "trade_name": "UAT Client A", "client_type": "Company", "pan": "AABCU1111A", "gstin": "33AABCU1111A1Z5", "tan": "CHEU12345A", "cin_llpin": "U12345TN2025PTC111111", "msme_no": "UDYAM-TN-00-0001111", "iec_code": "IEC1111111", "contact_person_name": "UAT Client A Contact", "mobile": "9000000001", "email": "uat.client@vavalam.com", "alternate_email": "support@vavalam.com", "city": "Chennai", "state": "Tamil Nadu", "pincode": "600001", "country": "India", "status": "active", "client_category": "Regular", "risk_category": "Medium", "onboarding_date": date.today(), "gst_applicable": True, "income_tax_applicable": True, "tds_applicable": True, "roc_applicable": True, "audit_applicable": True, "is_active": True, }, ) client_b, _ = get_or_create( Client, {"tenant_id": tenant_b.id, "client_code": "UAT-CL-B"}, { "branch_id": branch_b.id, "partner_id": tenant_b_admin.id, "default_review_partner_user_id": tenant_b_admin.id, "portal_user_id": client2_user.id, "engagement_mode": "hybrid", "client_name": "UAT Client B Pvt Ltd", "trade_name": "UAT Client B", "client_type": "Company", "pan": "AABCU2222A", "gstin": "33AABCU2222A1Z5", "email": "uat.client2@vavalam.com", "mobile": "9000000002", "city": "Chennai", "state": "Tamil Nadu", "pincode": "600002", "country": "India", "status": "active", "client_category": "Regular", "risk_category": "Low", "onboarding_date": date.today(), "gst_applicable": True, "income_tax_applicable": True, "is_active": True, }, ) ids["CLIENT_A_ID"] = client_a.id ids["CLIENT_B_ID"] = client_b.id # Service catalogue and task templates gst_cat = ensure_service_category("GST", "GST Compliance", 10) tax_cat = ensure_service_category("IT", "Income Tax", 20) audit_cat = ensure_service_category("AUDIT", "Audit & Assurance", 30) roc_cat = ensure_service_category("ROC", "ROC / MCA", 40) gst3b = ensure_catalogue("GST-GSTR3B-M", "GSTR-3B Monthly Filing", gst_cat, "monthly", "non_assurance", 10) gstr1 = ensure_catalogue("GST-GSTR1-M", "GSTR-1 Monthly Filing", gst_cat, "monthly", "non_assurance", 11) itr = ensure_catalogue("IT-ITR-COMP", "Company Income Tax Return", tax_cat, "yearly", "non_assurance", 20) audit = ensure_catalogue("AUD-TAX-AUDIT", "Tax Audit", audit_cat, "yearly", "assurance", 30) roc = ensure_catalogue("ROC-AOC4-MGT7", "AOC-4 and MGT-7 Filing", roc_cat, "yearly", "non_assurance", 40) ids["SERVICE_A_ID"] = gst3b.id task_specs = { gst3b: [(1, "Collect GST data", "Staff", False, "GST_DATA"), (2, "Prepare GSTR-3B working", "Staff", True, "GST_WORKING"), (3, "Partner review and filing", "Partner", True, "GST_RETURN")], gstr1: [(1, "Collect sales register", "Staff", False, "SALES_REGISTER"), (2, "Prepare GSTR-1", "Staff", True, "GST_RETURN")], itr: [(1, "Collect financial statements", "Staff", False, "FINANCIALS"), (2, "Prepare computation", "Staff", True, "COMPUTATION"), (3, "Review and file return", "Partner", True, "IT_RETURN")], audit: [(1, "Plan audit", "Partner", True, "AUDIT_PLAN"), (2, "Execute audit checklist", "Staff", True, "AUDIT_WORKING"), (3, "Issue audit report", "Partner", True, "AUDIT_REPORT")], roc: [(1, "Collect signed financials", "Staff", False, "SIGNED_FS"), (2, "Prepare forms", "Staff", True, "ROC_FORM"), (3, "File MCA forms", "Partner", True, "ROC_CHALLAN")], } firm_tasks_a = {} firm_tasks_b = {} for svc, specs in task_specs.items(): enable_firm_service(tenant_a, branch_a, svc, firm_admin) enable_firm_service(tenant_b, branch_b, svc, tenant_b_admin) for seq, task_name, role, review, doc_type in specs: ensure_default_task(svc, seq, task_name, role, review) fta = ensure_firm_task(tenant_a, svc, seq, task_name, role, firm_admin, review) ftb = ensure_firm_task(tenant_b, svc, seq, task_name, role, tenant_b_admin, review) ensure_doc_requirement(tenant_a, svc, fta, f"{doc_type} Document", doc_type) ensure_doc_requirement(tenant_b, svc, ftb, f"{doc_type} Document", doc_type) firm_tasks_a[(svc.service_code, seq)] = fta firm_tasks_b[(svc.service_code, seq)] = ftb gst3b_rule = ensure_due_rule(gst3b, "GSTR-3B due date", "monthly", 20, None, 1) gstr1_rule = ensure_due_rule(gstr1, "GSTR-1 due date", "monthly", 11, None, 1) itr_rule = ensure_due_rule(itr, "Company ITR due date", "yearly", 31, 10, 0) audit_rule = ensure_due_rule(audit, "Tax audit due date", "yearly", 30, 9, 0) roc_rule = ensure_due_rule(roc, "ROC annual filing due date", "yearly", 29, 10, 0) sel_gst_a = enable_firm_service(tenant_a, branch_a, gst3b, firm_admin) sel_gst_b = enable_firm_service(tenant_b, branch_b, gst3b, tenant_b_admin) sub_a = ensure_subscription(tenant_a, branch_a, client_a, gst3b, sel_gst_a, partner, manager, staff, gst3b_rule) sub_b = ensure_subscription(tenant_b, branch_b, client_b, gst3b, sel_gst_b, tenant_b_admin, tenant_b_admin, branch2staff, gst3b_rule) ids["ENGAGEMENT_A_ID"] = sub_a.id ids["ENGAGEMENT_B_ID"] = sub_b.id task_a = ensure_task_instance(tenant_a, branch_a, client_a, sub_a, gst3b, firm_tasks_a[("GST-GSTR3B-M", 1)], staff, 5) ensure_task_instance(tenant_a, branch_a, client_a, sub_a, gst3b, firm_tasks_a[("GST-GSTR3B-M", 2)], staff, 7) ensure_task_instance(tenant_a, branch_a, client_a, sub_a, gst3b, firm_tasks_a[("GST-GSTR3B-M", 3)], partner, 9) ensure_task_instance(tenant_b, branch_b, client_b, sub_b, gst3b, firm_tasks_b[("GST-GSTR3B-M", 1)], branch2staff, 5) # Notice case if NoticeCase is not None: nc, _ = get_or_create( NoticeCase, {"reference_no": "UAT-NOTICE-001"}, { "tenant_id": tenant_a.id, "branch_id": branch_a.id, "client_id": client_a.id, "case_code": "UAT-NC-A-001", "department": "GST", "case_type": "Notice", "title": "UAT GST Notice", "status": "Open", "notice_date": date.today(), "due_date": date.today() + timedelta(days=15), "issue_summary": "Seed notice case for Playwright tests", }, ) ids["NOTICE_CASE_A_ID"] = nc.id # Document storage agent node + sample engagement document/version/job secret_hash = hashlib.sha256(LSA_SECRET.encode("utf-8")).hexdigest() node_a, _ = get_or_create( BranchStorageNode, {"tenant_id": tenant_a.id, "branch_id": branch_a.id, "node_code": "UAT-LSA-A"}, { "node_name": "UAT Branch A Local Storage Agent", "connector_url": None, "secret_key_hash": secret_hash, "storage_root_path": "D:/UAT/BranchA/Documents", "storage_mode": "pull_jobs", "status": "active", "is_active": True, "quota_limit_bytes": 1024 * 1024 * 1024, "used_storage_bytes": 0, "subscription_required": False, "created_by_user_id": firm_admin.id, }, ) ids["STORAGE_NODE_A_ID"] = node_a.id doc_a, _ = get_or_create( EngagementDocument, {"tenant_id": tenant_a.id, "engagement_id": sub_a.id, "document_code": "UAT-DOC-A-001"}, { "branch_id": branch_a.id, "client_id": client_a.id, "task_instance_id": task_a.id, "financial_year": ACTIVE_FY, "assessment_year": ASSESSMENT_YEAR, "document_type": "GST_DATA", "title": "UAT GST Working Document", "description": "Seeded document for UAT/VAPT tests", "current_version_no": 1, "status": "active", "is_deleted": False, "created_by_user_id": staff.id, "updated_by_user_id": staff.id, }, ) sample_hash = hashlib.sha256(b"UAT seeded document content\n").hexdigest() ver_a, _ = get_or_create( EngagementDocumentVersion, {"document_id": doc_a.id, "version_no": 1}, { "tenant_id": tenant_a.id, "branch_id": branch_a.id, "client_id": client_a.id, "engagement_id": sub_a.id, "original_filename": "uat-gst-working.txt", "stored_filename": "uat-gst-working-v1.txt", "content_type": "text/plain", "file_size_bytes": len(b"UAT seeded document content\n"), "file_hash_sha256": sample_hash, "storage_backend": "LOCAL_YEAR_WISE", "local_relative_path": "UAT-A/UAT-CL-A/2025-26/uat-gst-working-v1.txt", "storage_status": "stored", "remarks": "Seeded UAT version", "uploaded_by_user_id": staff.id, }, ) get_or_create( DocumentStorageJob, {"storage_node_id": node_a.id, "document_id": doc_a.id, "version_id": ver_a.id}, { "tenant_id": tenant_a.id, "branch_id": branch_a.id, "job_type": "store_version", "status": "pending", "priority": 5, "staging_relative_path": "staging/uat-gst-working-v1.txt", "target_relative_path": "UAT-A/UAT-CL-A/2025-26/uat-gst-working-v1.txt", "file_size_bytes": ver_a.file_size_bytes, "expected_hash_sha256": sample_hash, "attempts": 0, "created_by_user_id": staff.id, }, ) ids["TASK_DOCUMENT_A_ID"] = doc_a.id db.commit() print("\nSeed completed. Paste these into the Playwright .env:\n") for k, v in ids.items(): print(f"{k}={v}") print("\nUse these login emails in Playwright .env:") print("SYSTEM_ADMIN_EMAIL=uat.admin@vavalam.com") print("FIRM_ADMIN_EMAIL=uat.firmadmin@vavalam.com") print("PARTNER_EMAIL=uat.partner@vavalam.com") print("MANAGER_EMAIL=uat.manager@vavalam.com") print("STAFF_EMAIL=uat.staff@vavalam.com") print("STAFF2_EMAIL=uat.staff2@vavalam.com") print("BRANCH2_STAFF_EMAIL=uat.branch2staff@vavalam.com") print("CLIENT_EMAIL=uat.client@vavalam.com") print("CLIENT2_EMAIL=uat.client2@vavalam.com") print("CONSULTANT_EMAIL=uat.consultant@vavalam.com") print("INVITE_TEST_EMAIL=uat.invite@vavalam.com") print("RESET_TEST_EMAIL=uat.reset@vavalam.com") print("LOCKED_USER_EMAIL=uat.lockeduser@vavalam.com") print("TENANT_B_ADMIN_EMAIL=uat.firmadminb@vavalam.com") print(f"\nSeed password for all UAT ERP users: {PASSWORD}") print(f"Local storage agent test secret: {LSA_SECRET}") print("\nNOTE: Create mailbox uat.firmadminb@vavalam.com if OTP login is required for Tenant B Admin tests.") except Exception: db.rollback() raise finally: db.close() if __name__ == "__main__": main()