Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
arrr-erp/scripts/seed_uat_data_v2_5_1_FIX5.py
T

760 lines
38 KiB
Python
Raw Permalink Blame History

"""
Audit Firm ERP -- UAT seed extension for Playwright v2.5.1 Deep SQLite suite.
Purpose:
Extend the existing core UAT seed with supporting records for the new v2.5.1
deep tests: email integration, marketplace leads, platform billing, RBAC,
audit logs, document security, HR, billing rules, notice-case rules and
financial-year locking.
Run inside ERP container, AFTER the core seed:
cd /app
PYTHONPATH=/app python scripts/seed_uat_data.py
PYTHONPATH=/app python scripts/seed_uat_data_v2_5_1.py
Safe/idempotent for UAT: it inserts or updates only UAT-coded records where the
matching tables/columns exist. If a module table is not present in this ERP build,
the script skips that area and prints a warning instead of failing.
Do NOT run against live production data.
"""
from __future__ import annotations
import hashlib
import json
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")
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")
SMTP_PASSWORD = os.getenv("UAT_SMTP_PASSWORD", "Pass@123##")
def fail(msg: str) -> None:
print(f"\n[v2.5.1 seed] 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(
"Could not import ERP DB modules. Run this from ERP project root inside the app container. "
f"Underlying 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())
report: list[str] = []
ids: dict[str, object] = {}
def norm(name: str) -> str:
return name.lower().replace("_", "").replace("-", "")
def find_table(*candidates: str):
# exact match first
for name in candidates:
if name in meta.tables:
return meta.tables[name]
# normalized exact match
normalized = {norm(t): t for t in table_names}
for name in candidates:
hit = normalized.get(norm(name))
if hit:
return meta.tables[hit]
# token contains fallback
for name in candidates:
tokens = [tok for tok in name.lower().replace("-", "_").split("_") if tok]
for t in table_names:
low = t.lower()
if all(tok in low for tok in tokens):
return meta.tables[t]
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 required_default_for_column(table, column_name: str):
"""Best-effort UAT defaults for reflected NOT NULL columns.
This keeps the seed resilient when the ERP schema has extra mandatory
fields that are not present in every build/version. Values are only
added when the column exists and is NOT NULL with no server/default.
"""
name = column_name.lower()
table_name = table.name.lower() if table is not None else ""
# Common foreign-key/context IDs
if name in {"tenant_id", "audit_firm_id"}:
return tenant_a_id
if name == "branch_id":
return branch_a_id
if name in {"created_by_user_id", "updated_by_user_id", "created_by_id", "updated_by_id"}:
return admin_id
if name in {"user_id", "employee_user_id"}:
return staff_id
if name == "client_id":
return client_a_id
if name in {"assigned_to_user_id", "owner_user_id"}:
return manager_id
# Platform billing mandatory classifiers
if name == "account_type":
return "audit_firm"
if name == "plan_type":
return "audit_firm"
if name == "subscription_type":
return "audit_firm"
if name == "invoice_type":
return "subscription"
if name == "payment_type":
return "invoice"
if name == "entity_type":
return "audit_firm"
# Email template mandatory columns
if name in {"body_template", "plain_body", "html_template", "body", "content"}:
return "Dear User, your UAT OTP is {{otp}}. This is for ERP UAT testing."
if name in {"subject_template", "subject"}:
return "UAT ERP Test Notification"
if name in {"template_name", "name", "title", "account_name", "display_name"}:
return f"UAT {table_name.replace('_', ' ').title()}"
if name == "template_type":
return "system"
# Common business defaults
if name == "currency":
return "INR"
if name == "billing_cycle":
return "monthly"
if name == "status":
return "active"
if name == "gst_registration_type":
return "regular"
if name in {"state", "place_of_supply"}:
return "Tamil Nadu"
if name in {"country"}:
return "India"
if name in {"address", "billing_address"}:
return "UAT Test Address"
if name in {"email", "billing_email"}:
return "support@vavalam.com"
if name in {"mobile", "phone", "contact_number"}:
return "9000000000"
if name.endswith("_code") or name in {"code", "account_code", "invoice_no", "payment_ref"}:
return f"UAT-{table_name[:10].upper()}"
if name.endswith("_no") or name.endswith("_number"):
return f"UAT-{table_name[:8].upper()}-001"
if name in {"description", "remarks", "notes"}:
return "Seeded for Playwright v2.5.1 UAT/VAPT testing"
# Date/time defaults
if name.endswith("_date") or name in {"start_date", "end_date", "due_date", "invoice_date", "payment_date"}:
return today()
if name.endswith("_at") or name in {"created_at", "updated_at"}:
return utcnow()
# Type based fallbacks, avoiding unknown FK IDs where possible
try:
pytype = table.c[column_name].type.python_type
except Exception:
pytype = str
if pytype is bool:
return False
if pytype is int:
if name.endswith("_id"):
return None
return 0
if pytype is float:
return 0.0
if pytype is date:
return today()
if pytype is datetime:
return utcnow()
return f"UAT {column_name.replace('_', ' ').title()}"
def add_required_defaults(table, data: dict) -> dict:
if table is None:
return data
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
value = required_default_for_column(table, col.name)
if value is not None:
out[col.name] = value
return valid(table, out)
def one(table, lookup: dict):
if table is None:
return None
data = valid(table, lookup)
if not data:
return None
stmt = select(table).where(and_(*[table.c[k] == v for k, v in data.items()]))
return db.execute(stmt).mappings().first()
def first(table, **lookup):
return one(table, lookup)
def first_by_any(table, lookups: list[dict]):
for lookup in lookups:
row = one(table, lookup)
if row:
return row
return None
def derive_lookup(table, lookup: dict, defaults: dict | None = None) -> dict:
"""Build a valid idempotent lookup even when ERP column names vary.
Earlier versions skipped rows when a table had no exact code column
such as lead_code/plan_code/event_code. This helper uses a safe
existing unique-ish column combination instead, so the seed can create
rows in more ERP schema variants.
"""
if table is None:
return {}
merged = {**(defaults or {}), **(lookup or {})}
merged_v = valid(table, merged)
if not merged_v:
return {}
cols = set(table.c.keys())
preferred_sets = [
["tenant_id", "branch_id", "template_code"],
["tenant_id", "template_code"],
["tenant_id", "branch_id", "lead_code"],
["tenant_id", "lead_code"],
["tenant_id", "account_code"],
["tenant_id", "subscription_code"],
["tenant_id", "invoice_no"],
["tenant_id", "invoice_number"],
["tenant_id", "payment_ref"],
["tenant_id", "message_id"],
["tenant_id", "event_code"],
["tenant_id", "action", "entity_type"],
["tenant_id", "email"],
["tenant_id", "from_email", "subject"],
["tenant_id", "name"],
["tenant_id", "template_name"],
["tenant_id", "title"],
["plan_code"],
["code"],
["name"],
["plan_name"],
["title"],
]
for keys in preferred_sets:
if all(k in cols and k in merged_v and merged_v[k] is not None for k in keys):
return {k: merged_v[k] for k in keys}
# Last resort: use tenant_id plus the first available stable text/code column.
text_like = [
"code", "name", "title", "email", "subject", "reference_no", "invoice_no",
"invoice_number", "payment_ref", "message_id", "action", "status",
]
out = {}
if "tenant_id" in cols and merged_v.get("tenant_id") is not None:
out["tenant_id"] = merged_v["tenant_id"]
for key in text_like:
if key in cols and merged_v.get(key) is not None:
out[key] = merged_v[key]
break
if out:
return out
# Absolute fallback: first non-null non-PK column from data.
for key, value in merged_v.items():
if value is not None and key in cols and not table.c[key].primary_key:
return {key: value}
return {}
def upsert(table, lookup: dict, defaults: dict | None = None, label: str = "record"):
if table is None:
report.append(f"SKIP {label}: table missing")
return None
defaults_v = valid(table, defaults or {})
lookup_v = valid(table, lookup)
if not lookup_v:
lookup_v = derive_lookup(table, lookup, defaults_v)
if not lookup_v:
report.append(f"SKIP {label}: no matching lookup columns in {table.name}")
return None
row = one(table, lookup_v)
if row:
if defaults_v:
db.execute(update(table).where(and_(*[table.c[k] == v for k, v in lookup_v.items()])).values(**defaults_v))
db.flush()
return one(table, lookup_v)
data = add_required_defaults(table, {**lookup_v, **defaults_v})
try:
with db.begin_nested():
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:
# Keep the seed moving when optional/deep-test modules have schema differences.
# The related Playwright cases will then show a module/seed gap instead of blocking all seed data.
report.append(f"SKIP {label}: insert failed in {table.name}: {exc.__class__.__name__}: {str(exc).splitlines()[0]}")
return None
def update_existing(table, lookup: dict, values: dict, label: str):
if table is None:
report.append(f"SKIP {label}: table missing")
return None
lookup_v = valid(table, lookup)
values_v = valid(table, values)
if not lookup_v or not values_v:
report.append(f"SKIP {label}: no matching columns in {table.name}")
return None
db.execute(update(table).where(and_(*[table.c[k] == v for k, v in lookup_v.items()])).values(**values_v))
db.flush()
return one(table, lookup_v)
def val(row, key: str, default=None):
return row[key] if row and key in row else default
# Core table references from earlier seed
tenants = find_table("tenants", "tenant")
branches = find_table("branches", "branch")
users = find_table("users", "user")
clients = find_table("clients", "client")
employees = find_table("employees", "employee")
financial_years = find_table("financial_years", "financial_year")
documents = find_table("engagement_documents", "documents", "client_documents")
document_versions = find_table("engagement_document_versions", "document_versions")
storage_nodes = find_table("branch_storage_nodes", "storage_nodes")
storage_jobs = find_table("document_storage_jobs", "storage_jobs")
notice_cases = find_table("notice_cases", "cases")
subscriptions = find_table("client_service_subscriptions", "subscriptions", "engagements")
task_instances = find_table("client_service_task_instances", "task_instances", "tasks")
services = find_table("service_catalogue", "service_catalogues", "services")
roles = find_table("roles", "rbac_roles")
permissions = find_table("permissions", "rbac_permissions")
role_permissions = find_table("role_permissions", "rbac_role_permissions")
tenant_a = first_by_any(tenants, [{"code": "UAT-A"}, {"tenant_code": "UAT-A"}, {"name": "UAT Tenant A"}])
tenant_b = first_by_any(tenants, [{"code": "UAT-B"}, {"tenant_code": "UAT-B"}, {"name": "UAT Tenant B"}])
branch_a = first_by_any(branches, [{"code": "UAT-BA"}, {"branch_code": "UAT-BA"}, {"name": "UAT Branch A"}])
branch_b = first_by_any(branches, [{"code": "UAT-BB"}, {"branch_code": "UAT-BB"}, {"name": "UAT Branch B"}])
firm_admin = first(users, email=f"uat.firmadmin@{UAT_DOMAIN}")
system_admin = first(users, email=f"uat.admin@{UAT_DOMAIN}")
partner = first(users, email=f"uat.partner@{UAT_DOMAIN}")
manager = first(users, email=f"uat.manager@{UAT_DOMAIN}")
staff = first(users, email=f"uat.staff@{UAT_DOMAIN}")
staff2 = first(users, email=f"uat.staff2@{UAT_DOMAIN}")
client_user = first(users, email=f"uat.client@{UAT_DOMAIN}")
consultant = first(users, email=f"uat.consultant@{UAT_DOMAIN}")
client_a = first_by_any(clients, [{"client_code": "UAT-CL-A"}, {"code": "UAT-CL-A"}, {"client_name": "UAT Client A Pvt Ltd"}])
client_b = first_by_any(clients, [{"client_code": "UAT-CL-B"}, {"code": "UAT-CL-B"}, {"client_name": "UAT Client B Pvt Ltd"}])
service_a = first_by_any(services, [{"service_code": "GST-GSTR3B-M"}, {"code": "GST-GSTR3B-M"}, {"service_name": "GSTR-3B Monthly Filing"}])
sub_a = first_by_any(subscriptions, [{"financial_year": ACTIVE_FY, "client_id": val(client_a, "id")}, {"client_id": val(client_a, "id")}])
task_a = first_by_any(task_instances, [{"client_id": val(client_a, "id")}, {"assigned_to_user_id": val(staff, "id")}])
if not tenant_a or not branch_a or not firm_admin:
fail("Core UAT seed appears missing. Run seed_uat_data.py first.")
tenant_a_id = val(tenant_a, "id")
tenant_b_id = val(tenant_b, "id")
branch_a_id = val(branch_a, "id")
branch_b_id = val(branch_b, "id")
firm_admin_id = val(firm_admin, "id")
system_admin_id = val(system_admin, "id", firm_admin_id)
partner_id = val(partner, "id", firm_admin_id)
manager_id = val(manager, "id", firm_admin_id)
staff_id = val(staff, "id", firm_admin_id)
staff2_id = val(staff2, "id", staff_id)
client_user_id = val(client_user, "id", firm_admin_id)
consultant_id = val(consultant, "id", firm_admin_id)
client_a_id = val(client_a, "id")
client_b_id = val(client_b, "id")
service_a_id = val(service_a, "id")
sub_a_id = val(sub_a, "id")
task_a_id = val(task_a, "id")
common_scope = {
"tenant_id": tenant_a_id,
"branch_id": branch_a_id,
"created_by_user_id": firm_admin_id,
"updated_by_user_id": firm_admin_id,
"created_by_id": firm_admin_id,
"updated_by_id": firm_admin_id,
"created_at": utcnow(),
"updated_at": utcnow(),
"is_active": True,
}
# 1. Email integration seed
email_settings = find_table("email_settings", "firm_email_settings", "smtp_settings", "mail_settings")
smtp = upsert(email_settings, {"tenant_id": tenant_a_id, "setting_code": "UAT-SMTP-A"}, {
**common_scope,
"name": "UAT SMTP Settings",
"smtp_host": "mail.vavalam.com",
"smtp_port": 587,
"smtp_username": "no-reply@vavalam.com",
"smtp_password": SMTP_PASSWORD,
"from_email": "no-reply@vavalam.com",
"from_name": "ARRR ERP UAT",
"use_tls": True,
"imap_host": "mail.vavalam.com",
"imap_port": 993,
"imap_username": f"uat.firmadmin@{UAT_DOMAIN}",
"imap_password": SMTP_PASSWORD,
"status": "active",
"is_default": True,
}, "email settings")
ids["EMAIL_SETTINGS_ID"] = val(smtp, "id")
email_templates = find_table("email_templates", "mail_templates")
for code, subject, body in [
("UAT_OTP", "Your ERP OTP", "Your OTP is {{ otp }}"),
("UAT_INVITE", "ERP Invitation", "Please accept invitation: {{ invite_url }}"),
("UAT_NOTICE", "Notice update", "Notice {{ reference_no }} requires attention"),
]:
template_label = code.replace("_", " ")
upsert(email_templates, {"tenant_id": tenant_a_id, "template_code": code}, {
**common_scope,
# Different ERP builds use different mandatory/template column names.
"name": template_label,
"template_name": template_label,
"title": template_label,
"subject": subject,
"email_subject": subject,
"subject_template": subject,
"template_subject": subject,
"body": body,
"body_text": body,
"body_html": body,
"html_body": body,
"template_body": body,
"body_template": body,
"plain_body": body,
"html_template": body,
"content": body,
"template_content": body,
"template_type": "system",
"category": "system",
"is_active": True,
}, f"email template {code}")
email_queue = find_table("email_queue", "email_queues", "queued_emails", "mail_queue", "outgoing_emails", "email_outbox")
upsert(email_queue, {"tenant_id": tenant_a_id, "queue_code": "UAT-EMAIL-Q-001"}, {
**common_scope, "to_email": f"uat.client@{UAT_DOMAIN}", "from_email": "no-reply@vavalam.com",
"subject": "UAT queued email", "body": "Queued email for UAT", "status": "pending",
"priority": 5, "scheduled_at": utcnow(), "attempts": 0,
}, "email queue")
email_logs = find_table("email_logs", "mail_logs", "email_delivery_logs")
upsert(email_logs, {"tenant_id": tenant_a_id, "message_id": "UAT-EMAIL-LOG-001"}, {
**common_scope, "to_email": f"uat.client@{UAT_DOMAIN}", "from_email": "no-reply@vavalam.com",
"subject": "UAT delivered email", "status": "sent", "sent_at": utcnow(), "provider_response": "UAT seeded",
}, "email log")
inbox = find_table("email_inbox_messages", "email_inbox", "email_inboxes", "inbound_emails", "incoming_emails", "mail_inbox", "email_messages")
upsert(inbox, {"tenant_id": tenant_a_id, "message_id": "UAT-INBOX-001"}, {
**common_scope, "from_email": f"uat.client@{UAT_DOMAIN}", "to_email": "support@vavalam.com",
"subject": "UAT client document submission", "body": "Please map this email to the UAT client.",
"received_at": utcnow(), "status": "unmapped", "client_id": client_a_id,
}, "email inbox")
# 2. Marketplace / public lead seed
leads = find_table("marketplace_leads", "leads", "public_leads")
lead1 = upsert(leads, {"tenant_id": tenant_a_id, "lead_code": "UAT-LEAD-001", "email": "lead@example.com", "name": "UAT Marketplace Lead"}, {
**common_scope, "source": "public_form", "lead_type": "compliance", "name": "UAT Marketplace Lead",
"company_name": "UAT Lead Pvt Ltd", "contact_name": "UAT Lead Contact", "email": "lead@example.com",
"mobile": "9000000101", "service_interest": "GST", "status": "new", "assigned_to_user_id": manager_id,
"notes": "Seeded marketplace lead for v2.5.1 tests",
}, "marketplace lead new")
lead2 = upsert(leads, {"tenant_id": tenant_a_id, "lead_code": "UAT-LEAD-CONVERTED", "email": "converted@example.com", "name": "UAT Converted Lead"}, {
**common_scope, "source": "referral", "name": "UAT Converted Lead", "email": "converted@example.com",
"mobile": "9000000102", "status": "converted", "assigned_to_user_id": partner_id, "converted_client_id": client_a_id,
}, "marketplace lead converted")
ids["MARKETPLACE_LEAD_ID"] = val(lead1, "id")
# 3. Platform billing seed
plans = find_table("platform_billing_plans", "platform_subscription_plans", "subscription_plans", "billing_subscription_plans", "billing_plans", "plans")
plan = upsert(plans, {
"plan_code": "UAT-PLAN-PRO",
"code": "UAT-PLAN-PRO",
"name": "UAT Professional Plan",
"plan_name": "UAT Professional Plan",
"title": "UAT Professional Plan",
}, {
**common_scope,
"plan_code": "UAT-PLAN-PRO", "code": "UAT-PLAN-PRO",
"name": "UAT Professional Plan", "plan_name": "UAT Professional Plan",
"title": "UAT Professional Plan", "display_name": "UAT Professional Plan",
"description": "Seed plan for platform billing tests",
"plan_type": "audit_firm", "account_type": "audit_firm", "subscription_type": "audit_firm",
"billing_cycle": "monthly", "currency": "INR", "status": "active",
"monthly_price": 9999, "annual_price": 99990, "price": 9999, "amount": 9999,
"base_amount": 9999, "tax_rate": 18, "gst_rate": 18,
"max_users": 25, "max_clients": 500, "is_active": True,
"created_at": utcnow(), "updated_at": utcnow(),
}, "platform plan")
accounts = find_table("platform_billing_accounts", "billing_accounts", "accounts")
account = upsert(accounts, {"tenant_id": tenant_a_id, "account_code": "UAT-PLAT-ACC-A"}, {
**common_scope, "account_type": "audit_firm", "account_name": "UAT Tenant A Billing Account",
"display_name": "UAT Tenant A Billing Account", "billing_email": "support@vavalam.com",
"gstin": "33AABCU1111A1Z5", "gst_registration_type": "regular",
"billing_address": "UAT Test Address", "state": "Tamil Nadu", "status": "active",
}, "platform billing account")
plat_sub = None
p_inv = None
if not val(plan, "id"):
report.append("SKIP platform subscription/invoice/payment: no platform billing plan row could be created/found; check platform billing plan table/columns")
elif not val(account, "id"):
report.append("SKIP platform subscription/invoice/payment: no platform billing account row could be created/found")
else:
plat_subs = find_table("platform_subscriptions", "platform_billing_subscriptions", "audit_firm_subscriptions", "subscriptions")
plat_sub = upsert(plat_subs, {"tenant_id": tenant_a_id, "subscription_code": "UAT-PLAT-SUB-A"}, {
**common_scope, "account_id": val(account, "id"), "plan_id": val(plan, "id"), "status": "active",
"start_date": today(), "end_date": today() + timedelta(days=365), "billing_cycle": "monthly",
}, "platform subscription")
if not val(plat_sub, "id"):
report.append("SKIP platform invoice/payment: no platform subscription row could be created/found")
else:
plat_invoices = find_table("platform_invoices", "platform_billing_invoices", "invoices")
p_inv = upsert(plat_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-PLAT-INV-001"}, {
**common_scope, "account_id": val(account, "id"), "subscription_id": val(plat_sub, "id"),
"invoice_date": today(), "due_date": today() + timedelta(days=15), "status": "issued",
"subtotal": 9999, "tax_amount": 1799.82, "total_amount": 11798.82, "balance_amount": 11798.82,
}, "platform invoice")
if val(p_inv, "id"):
payments = find_table("platform_payments", "invoice_payments", "payments")
upsert(payments, {"tenant_id": tenant_a_id, "payment_ref": "UAT-PLAT-PAY-001"}, {
**common_scope, "invoice_id": val(p_inv, "id"), "payment_date": today(), "amount": 5000,
"payment_mode": "bank_transfer", "status": "posted", "remarks": "Seed partial platform payment",
}, "platform payment")
ids["PLATFORM_INVOICE_ID"] = val(p_inv, "id")
# 4. RBAC seed
perm_codes = [
"uat.clients.read", "uat.documents.read", "uat.billing.read", "uat.email.manage", "uat.marketplace.manage",
]
for code in perm_codes:
upsert(permissions, {"code": code}, {
"name": code, "description": f"UAT permission {code}", "module": "UAT", "is_active": True,
"created_at": utcnow(), "updated_at": utcnow(),
}, f"permission {code}")
role = upsert(roles, {"name": "UAT Limited Tester"}, {
"description": "Seeded limited role for RBAC tests", "is_system": False, "is_active": True,
"tenant_id": tenant_a_id, "created_at": utcnow(), "updated_at": utcnow(),
}, "rbac role")
if role_permissions is not None and role is not None:
for code in perm_codes[:2]:
perm = first(permissions, code=code) if permissions is not None else None
upsert(role_permissions, {"role_id": val(role, "id"), "permission_id": val(perm, "id")}, {
"created_at": utcnow(), "created_by_user_id": firm_admin_id,
}, f"role permission {code}")
# 5. Audit logs seed
audit_logs = find_table("audit_logs", "activity_logs", "system_audit_logs")
for code, action, entity in [
("UAT-AUD-LOGIN-FAIL", "login_failed", "User"),
("UAT-AUD-DOC-DOWNLOAD", "document_download", "EngagementDocument"),
("UAT-AUD-PERM-DENIED", "permission_denied", "RBAC"),
]:
upsert(audit_logs, {"tenant_id": tenant_a_id, "event_code": code, "action": action, "entity_type": entity}, {
**common_scope, "event_code": code, "user_id": firm_admin_id, "action": action, "event_type": action,
"entity_type": entity, "entity_id": client_a_id or 1, "ip_address": "127.0.0.1",
"user_agent": "UAT seed", "details": json.dumps({"seed": True, "no_secret": True}),
"message": f"Seed audit event {action}",
}, f"audit log {code}")
# 6. Financial-year states
update_existing(financial_years, {"tenant_id": tenant_a_id, "year_code": PREVIOUS_FY}, {
"is_locked": True, "locked_at": utcnow(), "locked_by_user_id": firm_admin_id,
"lock_reason": "UAT locked FY for v2.5.1 tests",
}, "previous FY lock")
update_existing(financial_years, {"tenant_id": tenant_a_id, "year_code": ACTIVE_FY}, {
"is_locked": False, "is_current": True,
}, "active FY current/unlocked")
backups = find_table("financial_year_backups", "fy_backups", "backup_exports")
fy = first(financial_years, tenant_id=tenant_a_id, year_code=PREVIOUS_FY)
backup = upsert(backups, {"tenant_id": tenant_a_id, "backup_code": "UAT-FY-BACKUP-001"}, {
**common_scope, "financial_year_id": val(fy, "id"), "year_code": PREVIOUS_FY,
"status": "completed", "file_name": "uat-fy-backup-001.zip", "file_size_bytes": 1024,
"created_by_user_id": firm_admin_id,
}, "FY backup")
ids["YEAR_BACKUP_EXPORT_ID"] = val(backup, "id")
ids["LOCKED_FY"] = PREVIOUS_FY
# 7. Document security/deep seed
secret_hash = hashlib.sha256(LSA_SECRET.encode()).hexdigest()
node = upsert(storage_nodes, {"tenant_id": tenant_a_id, "branch_id": branch_a_id, "node_code": "UAT-LSA-SEC-A"}, {
**common_scope, "node_name": "UAT Security Storage Agent", "secret_key_hash": secret_hash,
"storage_root_path": "D:/UAT/Security", "storage_mode": "pull_jobs", "status": "active",
"quota_limit_bytes": 1024 * 1024 * 1024, "used_storage_bytes": 0,
}, "security storage node")
ids["STORAGE_NODE_SECURITY_ID"] = val(node, "id")
doc_sec = upsert(documents, {"tenant_id": tenant_a_id, "document_code": "UAT-DOC-SEC-001"}, {
**common_scope, "branch_id": branch_a_id, "client_id": client_a_id, "engagement_id": sub_a_id,
"task_instance_id": task_a_id, "financial_year": ACTIVE_FY, "assessment_year": ASSESSMENT_YEAR,
"document_type": "SECURITY", "title": "UAT Client Visible Document", "description": "Client visible UAT doc",
"current_version_no": 2, "status": "active", "is_deleted": False, "is_client_visible": True,
}, "client visible document")
ids["CLIENT_VISIBLE_DOCUMENT_ID"] = val(doc_sec, "id")
for version, filename in [(1, "uat-sec-v1.txt"), (2, "uat-sec-v2.txt")]:
content = f"UAT secure version {version}\n".encode()
upsert(document_versions, {"document_id": val(doc_sec, "id"), "version_no": version}, {
**common_scope, "tenant_id": tenant_a_id, "branch_id": branch_a_id, "client_id": client_a_id,
"engagement_id": sub_a_id, "original_filename": filename, "stored_filename": filename,
"content_type": "text/plain", "file_size_bytes": len(content),
"file_hash_sha256": hashlib.sha256(content).hexdigest(), "storage_backend": "LOCAL_YEAR_WISE",
"local_relative_path": f"UAT-A/UAT-CL-A/{ACTIVE_FY}/{filename}", "storage_status": "stored",
"uploaded_by_user_id": staff_id,
}, f"document version {version}")
doc_deleted = upsert(documents, {"tenant_id": tenant_a_id, "document_code": "UAT-DOC-DELETED-001"}, {
**common_scope, "branch_id": branch_a_id, "client_id": client_a_id, "engagement_id": sub_a_id,
"financial_year": ACTIVE_FY, "document_type": "SECURITY", "title": "UAT Deleted Document",
"status": "deleted", "is_deleted": True, "deleted_at": utcnow(), "deleted_by_user_id": firm_admin_id,
}, "deleted document")
ids["DELETED_DOCUMENT_ID"] = val(doc_deleted, "id")
storage_job = upsert(storage_jobs, {"tenant_id": tenant_a_id, "job_code": "UAT-STORAGE-JOB-SEC-001"}, {
**common_scope, "storage_node_id": val(node, "id"), "document_id": val(doc_sec, "id"),
"job_type": "store_version", "status": "pending", "priority": 1,
"staging_relative_path": "staging/uat-sec-v2.txt", "target_relative_path": f"UAT-A/UAT-CL-A/{ACTIVE_FY}/uat-sec-v2.txt",
"file_size_bytes": 21, "expected_hash_sha256": hashlib.sha256(b"UAT secure version 2\n").hexdigest(),
"attempts": 0,
}, "storage job security")
ids["STORAGE_JOB_SECURITY_ID"] = val(storage_job, "id")
# 8. HR seed
attendance = find_table("employee_attendance", "attendance", "employee_attendances")
upsert(attendance, {"tenant_id": tenant_a_id, "employee_id": val(first(employees, user_id=staff_id), "id"), "attendance_date": today()}, {
**common_scope, "user_id": staff_id, "status": "present", "check_in_time": utcnow(), "source": "seed",
"remarks": "Seed attendance for duplicate/manager tests",
}, "attendance")
leave_types = find_table("employee_leave_types", "leave_types", "hr_leave_types")
casual_leave = upsert(leave_types, {"tenant_id": tenant_a_id, "code": "CASUAL", "name": "Casual Leave"}, {
**common_scope, "code": "CASUAL", "leave_type_code": "CASUAL", "name": "Casual Leave",
"leave_type_name": "Casual Leave", "description": "UAT casual leave type",
"days_per_year": 12, "annual_quota": 12, "default_balance": 12, "is_paid": True,
"carry_forward_allowed": False, "status": "active", "is_active": True,
}, "leave type casual")
leave_type_id = val(casual_leave, "id")
leave_balances = find_table("employee_leave_balances", "leave_balances")
leave_balance = upsert(leave_balances, {"tenant_id": tenant_a_id, "user_id": staff_id, "leave_type": "casual", "leave_type_id": leave_type_id, "financial_year": ACTIVE_FY}, {
**common_scope, "user_id": staff_id, "leave_type": "casual", "leave_type_id": leave_type_id,
"employee_id": val(first(employees, user_id=staff_id), "id"), "opening_balance": 12,
"availed": 2, "balance": 10, "closing_balance": 10, "financial_year": ACTIVE_FY, "is_active": True,
}, "leave balance")
leave_requests = find_table("employee_leave_requests", "leave_requests", "leaves")
leave_req = upsert(leave_requests, {"tenant_id": tenant_a_id, "leave_code": "UAT-LEAVE-PENDING-001", "user_id": staff_id, "leave_type_id": leave_type_id}, {
**common_scope, "leave_code": "UAT-LEAVE-PENDING-001", "user_id": staff_id,
"employee_id": val(first(employees, user_id=staff_id), "id"),
"leave_type": "casual", "leave_type_id": leave_type_id,
"from_date": today() + timedelta(days=5), "to_date": today() + timedelta(days=6),
"days": 2, "total_days": 2, "status": "pending", "reason": "UAT seeded pending leave",
"approver_user_id": manager_id, "approved_by_user_id": manager_id, "balance_id": val(leave_balance, "id"),
}, "leave request pending")
ids["PENDING_LEAVE_ID"] = val(leave_req, "id")
payroll_runs = find_table("payroll_runs", "employee_payroll_runs", "payroll")
payroll = upsert(payroll_runs, {"tenant_id": tenant_a_id, "run_code": "UAT-PAYROLL-001"}, {
**common_scope, "financial_year": ACTIVE_FY, "period_month": today().month, "period_year": today().year,
"status": "generated", "gross_amount": 50000, "deduction_amount": 5000, "net_amount": 45000,
}, "payroll run")
payslips = find_table("payslips", "employee_payslips")
upsert(payslips, {"tenant_id": tenant_a_id, "payslip_code": "UAT-PAYSLIP-001"}, {
**common_scope, "payroll_run_id": val(payroll, "id"), "employee_id": val(first(employees, user_id=staff_id), "id"),
"user_id": staff_id, "gross_amount": 50000, "deduction_amount": 5000, "net_amount": 45000,
"status": "generated", "file_name": "uat-payslip-001.pdf",
}, "payslip")
# 9. Firm billing business seed
firm_invoices = find_table("billing_invoices", "firm_invoices", "client_invoices", "invoices")
draft_inv = upsert(firm_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-BILL-DRAFT-001"}, {
**common_scope, "client_id": client_a_id, "invoice_date": today(), "due_date": today() + timedelta(days=15),
"status": "draft", "subtotal": 10000, "tax_amount": 1800, "total_amount": 11800, "balance_amount": 11800,
"financial_year": ACTIVE_FY,
}, "billing draft invoice")
issued_inv = upsert(firm_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-BILL-ISSUED-001"}, {
**common_scope, "client_id": client_a_id, "invoice_date": today(), "due_date": today() + timedelta(days=15),
"status": "issued", "subtotal": 20000, "tax_amount": 3600, "total_amount": 23600, "balance_amount": 18600,
"financial_year": ACTIVE_FY,
}, "billing issued invoice")
cancel_inv = upsert(firm_invoices, {"tenant_id": tenant_a_id, "invoice_no": "UAT-BILL-CANCELLED-001"}, {
**common_scope, "client_id": client_a_id, "invoice_date": today(), "status": "cancelled",
"subtotal": 1000, "tax_amount": 180, "total_amount": 1180, "balance_amount": 0,
"financial_year": ACTIVE_FY,
}, "billing cancelled invoice")
firm_payments = find_table("billing_payments", "client_payments", "invoice_payments", "payments")
upsert(firm_payments, {"tenant_id": tenant_a_id, "payment_ref": "UAT-BILL-PAY-001"}, {
**common_scope, "invoice_id": val(issued_inv, "id"), "client_id": client_a_id,
"payment_date": today(), "amount": 5000, "payment_mode": "upi", "status": "posted",
}, "billing partial payment")
ids["BILLING_DRAFT_INVOICE_ID"] = val(draft_inv, "id")
ids["BILLING_ISSUED_INVOICE_ID"] = val(issued_inv, "id")
ids["BILLING_CANCELLED_INVOICE_ID"] = val(cancel_inv, "id")
# 10. Notice case business seed
if notice_cases is not None:
nc = upsert(notice_cases, {"tenant_id": tenant_a_id, "case_code": "UAT-NC-DEEP-001"}, {
**common_scope, "branch_id": branch_a_id, "client_id": client_a_id, "reference_no": "UAT-NOTICE-DEEP-001",
"department": "GST", "case_type": "Notice", "title": "UAT Deep GST Notice", "status": "Hearing Scheduled",
"notice_date": today() - timedelta(days=10), "hearing_date": today() + timedelta(days=10),
"order_date": None, "due_date": today() + timedelta(days=20), "assigned_to_user_id": staff_id,
"partner_user_id": partner_id, "internal_notes": "Internal note must not be client-visible",
"client_visible_notes": "Client visible notice update",
"issue_summary": "Seeded v2.5.1 notice case for business-rule testing",
}, "deep notice case")
ids["NOTICE_CASE_DEEP_ID"] = val(nc, "id")
nc_events = find_table("notice_case_events", "case_events", "notice_case_activities")
upsert(nc_events, {"tenant_id": tenant_a_id, "event_code": "UAT-NC-EVT-001"}, {
**common_scope, "case_id": val(nc, "id"), "event_type": "hearing_scheduled", "event_date": today(),
"notes": "Seeded hearing event", "is_client_visible": True, "created_by_user_id": staff_id,
}, "notice event")
db.commit()
print("\n[v2.5.1 seed] Completed seed extension.")
print("[v2.5.1 seed] Created/updated data for email, marketplace, platform billing, RBAC, audit, FY, documents, HR, billing and notice-case tests.")
print("\nUseful IDs for Playwright .env if needed:\n")
for k, v in ids.items():
if v is not None:
print(f"{k}={v}")
print("\nSkipped/missing-module notes:")
for line in report:
print(f"- {line}")
print("\nIf many rows were skipped, that means the corresponding ERP module/table is not present yet; tests for those modules should then fail/skip as ERP gaps, not seed failures.")
db.close()
if __name__ == "__main__":
try:
main()
except Exception:
raise
Reference in New Issue View Git Blame Copy Permalink