From d40f357607e705c981d9e32460ad37950a585f14 Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Mon, 22 Jun 2026 16:15:55 +0530 Subject: [PATCH] Update UAT seed data for real IMAP users and service coverage --- scripts/seed_uat_data.py | 689 +++++++++++++++++++++++++++++++++------ 1 file changed, 585 insertions(+), 104 deletions(-) diff --git a/scripts/seed_uat_data.py b/scripts/seed_uat_data.py index 1f054f1..223d12e 100644 --- a/scripts/seed_uat_data.py +++ b/scripts/seed_uat_data.py @@ -1,33 +1,34 @@ """ -Audit Firm ERP (Production_gitea) -- Playwright UAT/VAPT seed helper. +Audit Firm ERP -- UAT seed data for Playwright v2.4.1 IMAP. -CORRECTED for the Production_gitea schema: - - session factory: app.core.db.common.CommonSessionLocal - - User has no 'role' column; roles are assigned via the user_roles join table - - Real role names: "Firm Admin", "Partner", "Branch Manager", "Staff", - "Client", "Consultant" - - Client requires client_name (not 'name'); Tenant/Branch use code+name - - NoticeCase uses case_code + reference_no +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. -HOW TO RUN (inside the ERP container, from the project root, AFTER migrations): - alembic upgrade head - UAT_SEED_PASSWORD='YourStrongTestPass@123' python seed_uat_data.py +Run inside ERP container from project root after migrations: + cd /app + PYTHONPATH=/app UAT_SEED_PASSWORD='Password@123' python scripts/seed_uat_data.py -It is idempotent: re-running updates/re-uses existing UAT records rather than -duplicating them. It prints IDs to paste into the Playwright .env. - -SAFETY: run this against a UAT/staging database, NOT live production data. +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, timedelta +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") -# Map UAT logical role -> actual Role.name in Production_gitea ROLE_NAME = { + "system_admin": "System Admin", "firm_admin": "Firm Admin", "partner": "Partner", "manager": "Branch Manager", @@ -36,6 +37,16 @@ ROLE_NAME = { "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) @@ -43,21 +54,37 @@ def fail(msg: str) -> None: def main() -> None: - # --- imports from the live ERP project ------------------------------------ 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 + 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 - except Exception as exc: # pragma: no cover + 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 the ERP project root " - f"inside the app container. Underlying import error: {exc!r}" + "Could not import ERP modules. Run this from ERP project root inside the app container. " + f"Underlying import error: {exc!r}" ) - # NoticeCase is optional (only if module present) try: from app.modules.notice_cases.models import NoticeCase except Exception: @@ -66,121 +93,478 @@ def main() -> None: db = CommonSessionLocal() ids: dict[str, object] = {} - def get_or_create(Model, lookup: dict, defaults: dict): + 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} - # keep only real columns - valid = {c.name for c in Model.__table__.columns} - obj = Model(**{k: v for k, v in data.items() if k in valid}) + 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 = db.query(Role).filter(Role.name == role_name).first() + role = role_by_name(role_name) if not role: - print(f"[seed] WARNING: role '{role_name}' not found; " - f"run the app once so DEFAULT_ROLES are created. Skipping.") + print(f"[seed] WARNING: role '{role_name}' not found. Run/restart ERP startup seed first.") return - exists = ( - db.query(UserRole) - .filter(UserRole.user_id == user.id, UserRole.role_id == role.id) - .first() - ) + 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 ---------------------------------------------------------- - tenant_a, _ = get_or_create( - Tenant, {"code": "UAT-A"}, - {"name": "UAT Tenant A", "is_active": True}, - ) - tenant_b, _ = get_or_create( - Tenant, {"code": "UAT-B"}, - {"name": "UAT Tenant B", "is_active": True}, - ) + # 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 - - # --- Branches --------------------------------------------------------- - branch_a, _ = get_or_create( - Branch, {"code": "UAT-BA"}, - {"name": "UAT Branch A", "tenant_id": tenant_a.id, "is_active": True}, - ) - branch_b, _ = get_or_create( - Branch, {"code": "UAT-BB"}, - {"name": "UAT Branch B", "tenant_id": tenant_b.id, "is_active": True}, - ) ids["BRANCH_A_ID"] = branch_a.id ids["BRANCH_B_ID"] = branch_b.id - # --- Users (+ roles via user_roles) ----------------------------------- - users = [ - ("uat.firmadmin@tenant-a.test", "Firm Admin", "firm_admin", tenant_a, branch_a), - ("uat.partner@tenant-a.test", "Partner", "partner", tenant_a, branch_a), - ("uat.manager@tenant-a.test", "Manager", "manager", tenant_a, branch_a), - ("uat.staff@tenant-a.test", "Staff", "staff", tenant_a, branch_a), - ("uat.client@tenant-a.test", "Client", "client", tenant_a, branch_a), - ("uat.consultant@tenant-a.test","Consultant", "consultant", tenant_a, branch_a), - ("uat.firmadmin@tenant-b.test", "Firm Admin B","firm_admin", tenant_b, branch_b), - ] - for email, full_name, role_label, tenant, branch in users: - user, created = 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": True, - "is_locked": False, - "must_change_password": False, - }, - ) - if not created: - # refresh password on existing UAT users so logins stay known - user.password_hash = hash_password(PASSWORD) - db.flush() - assign_role(user, role_label) + 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 - # --- Clients (note: client_name is required) -------------------------- + # 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, {"client_code": "UAT-CL-A"}, + Client, + {"tenant_id": tenant_a.id, "client_code": "UAT-CL-A"}, { - "tenant_id": tenant_a.id, "branch_id": branch_a.id, - "client_name": "UAT Client A Pvt Ltd", "client_type": "Company", - "pan": "AABCU1111A", "gstin": "33AABCU1111A1Z5", - "email": "client.a@uat.test", "mobile": "9000000001", - "status": "active", "is_active": True, + "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, {"client_code": "UAT-CL-B"}, + Client, + {"tenant_id": tenant_b.id, "client_code": "UAT-CL-B"}, { - "tenant_id": tenant_b.id, "branch_id": branch_b.id, - "client_name": "UAT Client B Pvt Ltd", "client_type": "Company", - "pan": "AABCU2222A", "gstin": "33AABCU2222A1Z5", - "email": "client.b@uat.test", "mobile": "9000000002", - "status": "active", "is_active": True, + "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 - # --- Notice case (optional) ------------------------------------------- + # 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"}, + 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", + "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", @@ -188,12 +572,109 @@ def main() -> None: ) 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": 10 * 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(f"\nSeed password for all uat.* users: {PASSWORD}") + 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()