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