seed for new

This commit is contained in:
A R R R Associates
2026-06-30 12:14:40 +05:30
parent df4fd17211
commit a834dc0e9d
+562
View File
@@ -0,0 +1,562 @@
"""
Audit Firm ERP -- strict UAT seed patch for FY lock/backup Playwright tests.
Run inside ERP container after core UAT seed:
cd /app
PYTHONPATH=/app python scripts/seed_uat_fy_lock_backup_strict.py
Purpose:
Creates deterministic UAT records needed by tests/fy-lock-backup.spec.js without
weakening test assertions or changing ERP features.
Scope:
- Financial years: 2025-26 current/open and 2024-25 locked for Tenant A
- Tenant/branch B scenario rows where base seed exists
- Active FY and locked FY subscriptions/tasks
- Engagement document and permanent document records
- Notice case record
- Billing invoice/payment records
The script is idempotent and only upserts UAT-coded records. Intended for UAT/test
containers, not live production data.
"""
from __future__ import annotations
import hashlib
import json
import os
import sys
from datetime import date, datetime, timedelta, timezone
ACTIVE_FY = os.getenv("ACTIVE_FY", "2025-26")
LOCKED_FY = os.getenv("LOCKED_FY", "2024-25")
ACTIVE_AY = os.getenv("ACTIVE_AY", "2026-27")
LOCKED_AY = os.getenv("LOCKED_AY", "2025-26")
UAT_DOMAIN = os.getenv("UAT_EMAIL_DOMAIN", "vavalam.com")
def fail(msg: str) -> None:
print(f"\n[fy-strict-seed] ERROR: {msg}", file=sys.stderr)
sys.exit(1)
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def main() -> None:
try:
from sqlalchemy import MetaData, and_, inspect, select, update, text
from app.core.db.common import CommonSessionLocal
except Exception as exc:
fail(f"Run this from /app inside ERP container. Import error: {exc!r}")
db = CommonSessionLocal()
engine = db.get_bind()
meta = MetaData()
meta.reflect(bind=engine)
inspector = inspect(engine)
table_names = set(inspector.get_table_names())
notes: list[str] = []
def norm(name: str) -> str:
return name.lower().replace("_", "").replace("-", "")
def table(*names: str):
for name in names:
if name in meta.tables:
return meta.tables[name]
lookup = {norm(t): t for t in table_names}
for name in names:
hit = lookup.get(norm(name))
if hit:
return meta.tables[hit]
return None
def valid(t, data: dict) -> dict:
if t is None:
return {}
cols = set(t.c.keys())
return {k: v for k, v in data.items() if k in cols and v is not None}
def where(t, lookup: dict):
lookup = valid(t, lookup)
if not lookup:
return None
return and_(*[t.c[k] == v for k, v in lookup.items()])
def one(t, lookup: dict):
if t is None:
return None
clause = where(t, lookup)
if clause is None:
return None
return db.execute(select(t).where(clause)).mappings().first()
def first(t, *lookups: dict):
if t is None:
return None
for lookup in lookups:
row = one(t, lookup)
if row:
return row
return db.execute(select(t).limit(1)).mappings().first()
def scalar(sql: str):
try:
return db.execute(text(sql)).scalar()
except Exception:
return None
def py_default(t, col_name: str):
name = col_name.lower()
if name.endswith("_at_utc") or name.endswith("_at"):
return utcnow()
if name.endswith("_date"):
return date.today()
if name in {"status", "export_status"}:
return "active"
if name in {"is_active"}:
return True
if name in {"is_deleted", "is_locked", "is_current"}:
return False
try:
pytype = t.c[col_name].type.python_type
except Exception:
pytype = str
if pytype is bool:
return False
if pytype is int:
return 0
if pytype is float:
return 0.0
if pytype is date:
return date.today()
if pytype is datetime:
return utcnow()
return f"UAT {col_name.replace('_', ' ').title()}"
def add_required_defaults(t, data: dict) -> dict:
out = dict(data)
for col in t.c:
if col.name in out:
continue
if col.primary_key or col.nullable or col.default is not None or col.server_default is not None:
continue
out[col.name] = py_default(t, col.name)
return valid(t, out)
def upsert(t, lookup: dict, defaults: dict, label: str):
if t is None:
notes.append(f"SKIP {label}: table missing")
return None
lookup_v = valid(t, lookup)
if not lookup_v:
notes.append(f"SKIP {label}: lookup columns missing in {t.name}")
return None
row = one(t, lookup_v)
values = valid(t, defaults)
if row:
if values:
db.execute(update(t).where(where(t, lookup_v)).values(**values))
db.flush()
return one(t, lookup_v)
data = add_required_defaults(t, {**lookup_v, **values})
try:
result = db.execute(t.insert().values(**data))
db.flush()
pk = result.inserted_primary_key[0] if result.inserted_primary_key else None
if pk and "id" in t.c:
return one(t, {"id": pk})
return one(t, lookup_v)
except Exception as exc:
db.rollback()
notes.append(f"SKIP {label}: insert failed in {t.name}: {exc.__class__.__name__}: {str(exc).splitlines()[0]}")
return one(t, lookup_v)
def v(row, key: str, default=None):
return row[key] if row is not None and key in row else default
tenants = table("tenants")
branches = table("branches")
users = table("users")
clients = table("clients")
service_catalogues = table("service_catalogues", "services")
financial_years = table("financial_years")
subscriptions = table("client_service_subscriptions")
tasks = table("client_service_task_instances")
engagement_documents = table("engagement_documents")
engagement_document_versions = table("engagement_document_versions")
permanent_docs = table("permanent_client_documents")
permanent_versions = table("permanent_client_document_versions")
notice_cases = table("notice_cases")
notice_docs = table("notice_case_documents")
invoices = table("billing_invoices")
payments = table("billing_payments")
tenant_a = first(tenants, {"code": "UAT-A"}, {"tenant_code": "UAT-A"}, {"name": "UAT Tenant A"})
tenant_b = first(tenants, {"code": "UAT-B"}, {"tenant_code": "UAT-B"}, {"name": "UAT Tenant B"}) or tenant_a
branch_a = first(branches, {"code": "UAT-BA"}, {"branch_code": "UAT-BA"}, {"name": "UAT Branch A"})
branch_b = first(branches, {"code": "UAT-BB"}, {"branch_code": "UAT-BB"}, {"name": "UAT Branch B"}) or branch_a
admin = first(users, {"email": f"uat.admin@{UAT_DOMAIN}"}, {"email": f"uat.firmadmin@{UAT_DOMAIN}"})
firm_admin = first(users, {"email": f"uat.firmadmin@{UAT_DOMAIN}"}) or admin
partner = first(users, {"email": f"uat.partner@{UAT_DOMAIN}"}) or firm_admin
manager = first(users, {"email": f"uat.manager@{UAT_DOMAIN}"}) or firm_admin
staff = first(users, {"email": f"uat.staff@{UAT_DOMAIN}"}) or firm_admin
branch_b_staff = first(users, {"email": f"uat.branch2staff@{UAT_DOMAIN}"}) or staff
client_user = first(users, {"email": f"uat.client@{UAT_DOMAIN}"})
if not tenant_a or not branch_a or not admin:
fail("Core UAT seed missing: tenant A, branch A, or admin user not found.")
tenant_a_id = v(tenant_a, "id")
tenant_b_id = v(tenant_b, "id")
branch_a_id = v(branch_a, "id")
branch_b_id = v(branch_b, "id")
admin_id = v(admin, "id")
partner_id = v(partner, "id")
manager_id = v(manager, "id")
staff_id = v(staff, "id")
branch_b_staff_id = v(branch_b_staff, "id")
# Ensure FY records required by the suite.
fy_active = upsert(financial_years, {"tenant_id": tenant_a_id, "year_code": ACTIVE_FY}, {
"assessment_year": ACTIVE_AY,
"start_date": date(int(ACTIVE_FY[:4]), 4, 1),
"end_date": date(int(ACTIVE_FY[:4]) + 1, 3, 31),
"is_current": True,
"is_locked": False,
"locked_at_utc": None,
"locked_by_user_id": None,
"updated_at_utc": utcnow(),
}, "active financial year")
fy_locked = upsert(financial_years, {"tenant_id": tenant_a_id, "year_code": LOCKED_FY}, {
"assessment_year": LOCKED_AY,
"start_date": date(int(LOCKED_FY[:4]), 4, 1),
"end_date": date(int(LOCKED_FY[:4]) + 1, 3, 31),
"is_current": False,
"is_locked": True,
"locked_at_utc": utcnow(),
"locked_by_user_id": admin_id,
"updated_at_utc": utcnow(),
}, "locked financial year")
if financial_years is not None and fy_active:
try:
db.execute(update(financial_years).where(
and_(financial_years.c.tenant_id == tenant_a_id, financial_years.c.year_code != ACTIVE_FY)
).values(is_current=False))
db.execute(update(financial_years).where(
and_(financial_years.c.tenant_id == tenant_a_id, financial_years.c.year_code == ACTIVE_FY)
).values(is_current=True, is_locked=False, locked_at_utc=None, locked_by_user_id=None))
db.flush()
except Exception as exc:
notes.append(f"Could not normalize current FY flag: {exc.__class__.__name__}: {exc}")
service = first(service_catalogues, {"service_code": "GST-GSTR3B-M"}, {"code": "GST-GSTR3B-M"}) or first(service_catalogues, {})
if not service:
fail("No service catalogue/service row available. Run service seed first.")
service_id = v(service, "id")
# Ensure Client A and Client B records where possible.
client_a = first(clients, {"client_code": "UAT-CL-A"}, {"code": "UAT-CL-A"}, {"client_name": "UAT Client A Pvt Ltd"})
client_a = client_a or upsert(clients, {"tenant_id": tenant_a_id, "client_code": "UAT-CL-A"}, {
"branch_id": branch_a_id,
"client_name": "UAT Client A Pvt Ltd",
"trade_name": "UAT Client A",
"pan": "ABCDE1234F",
"gstin": "33ABCDE1234F1Z5",
"email": f"uat.client@{UAT_DOMAIN}",
"mobile": "9000000004",
"status": "active",
"is_active": True,
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, "client A")
client_b = first(clients, {"client_code": "UAT-CL-B"}, {"code": "UAT-CL-B"}, {"client_name": "UAT Client B Pvt Ltd"})
client_b = client_b or upsert(clients, {"tenant_id": tenant_b_id, "client_code": "UAT-CL-B"}, {
"branch_id": branch_b_id,
"client_name": "UAT Client B Pvt Ltd",
"trade_name": "UAT Client B",
"pan": "ABCDE5678F",
"gstin": "33ABCDE5678F1Z5",
"email": f"uat.client2@{UAT_DOMAIN}",
"mobile": "9000000005",
"status": "active",
"is_active": True,
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, "client B")
client_a_id = v(client_a, "id")
client_b_id = v(client_b, "id", client_a_id)
# Subscriptions / engagements.
def sub_seed(label, tenant_id, branch_id, client_id, fy, ay, assignee_staff_id, locked=False):
return upsert(subscriptions, {"tenant_id": tenant_id, "client_id": client_id, "service_catalogue_id": service_id, "financial_year": fy}, {
"branch_id": branch_id,
"assessment_year": ay,
"assigned_partner_user_id": partner_id,
"assigned_manager_user_id": manager_id,
"assigned_staff_user_id": assignee_staff_id,
"review_partner_user_id": partner_id,
"engagement_type": "non_assurance",
"status": "active",
"is_active": True,
"is_locked": locked,
"locked_at_utc": utcnow() if locked else None,
"locked_by_user_id": admin_id if locked else None,
"start_date": date(int(fy[:4]), 4, 1),
"end_date": date(int(fy[:4]) + 1, 3, 31),
"original_due_date": date(int(fy[:4]) + 1, 4, 20),
"current_due_date": date(int(fy[:4]) + 1, 4, 20),
"remarks": f"Strict UAT seed engagement for {label}",
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, label)
sub_active = sub_seed("active FY subscription", tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, staff_id, False)
sub_locked = sub_seed("locked FY subscription", tenant_a_id, branch_a_id, client_a_id, LOCKED_FY, LOCKED_AY, staff_id, True)
sub_branch_b = sub_seed("branch B subscription", tenant_b_id, branch_b_id, client_b_id, ACTIVE_FY, ACTIVE_AY, branch_b_staff_id, False)
# Task instances.
def task_seed(label, sub, tenant_id, branch_id, client_id, fy, ay, task_name, assignee_id, status="pending", locked=False, seq=1):
return upsert(tasks, {"subscription_id": v(sub, "id"), "financial_year": fy, "task_name": task_name}, {
"tenant_id": tenant_id,
"branch_id": branch_id,
"client_id": client_id,
"service_catalogue_id": service_id,
"assessment_year": ay,
"description": f"Strict UAT seed task for {label}",
"sequence_no": seq,
"default_role_name": "Staff",
"assigned_to_user_id": assignee_id,
"internal_target_date": date.today() + timedelta(days=10),
"status": status,
"priority": "normal",
"is_active": True,
"is_locked": locked,
"locked_at_utc": utcnow() if locked else None,
"locked_by_user_id": admin_id if locked else None,
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, label)
task_active = task_seed("active FY task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT Active FY Task", staff_id, "pending", False, 10)
task_partner = task_seed("partner review task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT Partner Review Task", partner_id, "pending", False, 20)
task_in_progress = task_seed("in progress task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT In Progress Task", staff_id, "in_progress", False, 30)
task_completed = task_seed("completed task", sub_active, tenant_a_id, branch_a_id, client_a_id, ACTIVE_FY, ACTIVE_AY, "UAT Completed Task", staff_id, "completed", False, 40)
task_locked = task_seed("locked FY task", sub_locked, tenant_a_id, branch_a_id, client_a_id, LOCKED_FY, LOCKED_AY, "UAT Locked FY Task", staff_id, "pending", True, 50)
task_branch_b = task_seed("branch B active FY task", sub_branch_b, tenant_b_id, branch_b_id, client_b_id, ACTIVE_FY, ACTIVE_AY, "UAT Branch B Active FY Task", branch_b_staff_id, "pending", False, 60)
# Engagement document and version.
doc_content = b"Strict UAT FY engagement document\n"
eng_doc = upsert(engagement_documents, {"tenant_id": tenant_a_id, "engagement_id": v(sub_active, "id"), "document_code": "UAT-FY-ENG-DOC-001"}, {
"branch_id": branch_a_id,
"client_id": client_a_id,
"task_instance_id": v(task_active, "id"),
"financial_year": ACTIVE_FY,
"assessment_year": ACTIVE_AY,
"document_type": "GENERAL",
"title": "UAT FY Engagement Document",
"description": "Strict UAT seed engagement document",
"current_version_no": 1,
"status": "active",
"is_deleted": False,
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, "engagement document")
eng_doc_ver = upsert(engagement_document_versions, {"document_id": v(eng_doc, "id"), "version_no": 1}, {
"tenant_id": tenant_a_id,
"branch_id": branch_a_id,
"client_id": client_a_id,
"engagement_id": v(sub_active, "id"),
"original_filename": "uat-fy-engagement-document.txt",
"stored_filename": "uat-fy-engagement-document.txt",
"content_type": "text/plain",
"file_size_bytes": len(doc_content),
"file_hash_sha256": hashlib.sha256(doc_content).hexdigest(),
"storage_backend": "LOCAL_YEAR_WISE",
"local_relative_path": f"UAT-A/{ACTIVE_FY}/engagements/uat-fy-engagement-document.txt",
"storage_status": "stored",
"remarks": "Strict UAT seed engagement document version",
"uploaded_by_user_id": admin_id,
"uploaded_at_utc": utcnow(),
}, "engagement document version")
# Permanent document.
perm_doc = upsert(permanent_docs, {"tenant_id": tenant_a_id, "client_id": client_a_id, "document_code": "UAT-FY-PERM-DOC-001"}, {
"branch_id": branch_a_id,
"category": "KYC",
"title": "UAT FY Permanent Client Document",
"description": "Strict UAT seed permanent document",
"current_version_no": 1,
"status": "active",
"is_deleted": False,
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, "permanent client document")
perm_content = b"Strict UAT permanent document\n"
perm_doc_ver = upsert(permanent_versions, {"document_id": v(perm_doc, "id"), "version_no": 1}, {
"tenant_id": tenant_a_id,
"branch_id": branch_a_id,
"client_id": client_a_id,
"original_filename": "uat-fy-permanent-document.txt",
"stored_filename": "uat-fy-permanent-document.txt",
"content_type": "text/plain",
"file_size_bytes": len(perm_content),
"file_hash_sha256": hashlib.sha256(perm_content).hexdigest(),
"storage_backend": "LOCAL_PERMANENT",
"local_relative_path": "UAT-A/UAT-CL-A/permanent/uat-fy-permanent-document.txt",
"storage_status": "stored",
"remarks": "Strict UAT seed permanent document version",
"uploaded_by_user_id": admin_id,
"uploaded_at_utc": utcnow(),
}, "permanent client document version")
# Notice case and document.
notice_case = upsert(notice_cases, {"tenant_id": tenant_a_id, "case_code": "UAT-FY-NC-001"}, {
"branch_id": branch_a_id,
"client_id": client_a_id,
"engagement_id": v(sub_active, "id"),
"service_catalogue_id": service_id,
"financial_year": ACTIVE_FY,
"assessment_year": ACTIVE_AY,
"reference_no": "UAT-FY-NOTICE-001",
"notice_no": "UAT-FY-NOTICE-001",
"case_title": "UAT FY Notice Case",
"title": "UAT FY Notice Case",
"department": "GST",
"case_type": "notice",
"status": "open",
"priority": "normal",
"notice_date": date.today(),
"due_date": date.today() + timedelta(days=30),
"description": "Strict UAT seed notice case",
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
"is_active": True,
}, "notice case")
notice_doc = upsert(notice_docs, {"tenant_id": tenant_a_id, "case_id": v(notice_case, "id"), "title": "UAT FY Notice Case Document"}, {
"branch_id": branch_a_id,
"document_type": "GENERAL",
"description": "Strict UAT seed notice case document",
"original_filename": "uat-fy-notice-document.txt",
"stored_filename": "uat-fy-notice-document.txt",
"content_type": "text/plain",
"file_size_bytes": 29,
"local_relative_path": "notice_cases/uat-fy-notice-document.txt",
"version_no": 1,
"status": "active",
"is_deleted": False,
"uploaded_by_user_id": admin_id,
"uploaded_at_utc": utcnow(),
}, "notice case document")
# Billing invoice and payment.
invoice = upsert(invoices, {"tenant_id": tenant_a_id, "invoice_no": f"UAT-FY-INV-{ACTIVE_FY}"}, {
"branch_id": branch_a_id,
"client_id": client_a_id,
"financial_year": ACTIVE_FY,
"invoice_date": date.today(),
"billing_period_from": date(int(ACTIVE_FY[:4]), 4, 1),
"billing_period_to": date(int(ACTIVE_FY[:4]) + 1, 3, 31),
"status": "ISSUED",
"subtotal_amount": 10000.00,
"taxable_amount": 10000.00,
"cgst_amount": 900.00,
"sgst_amount": 900.00,
"igst_amount": 0.00,
"total_tax_amount": 1800.00,
"total_amount": 11800.00,
"amount_paid": 1000.00,
"balance_amount": 10800.00,
"notes": "Strict UAT seed invoice for FY tests",
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, "billing invoice")
payment = upsert(payments, {"tenant_id": tenant_a_id, "receipt_no": f"UAT-FY-RCPT-{ACTIVE_FY}"}, {
"branch_id": branch_a_id,
"client_id": client_a_id,
"invoice_id": v(invoice, "id"),
"financial_year": ACTIVE_FY,
"payment_date": date.today(),
"amount_received": 1000.00,
"tds_deducted": 0.00,
"mode": "BANK_TRANSFER",
"reference_no": "UAT-FY-SEED",
"notes": "Strict UAT seed payment for FY tests",
"created_by_user_id": admin_id,
"updated_by_user_id": admin_id,
"created_at_utc": utcnow(),
"updated_at_utc": utcnow(),
}, "billing payment")
db.commit()
env_ids = {
"TENANT_A_ID": tenant_a_id,
"TENANT_B_ID": tenant_b_id,
"BRANCH_A_ID": branch_a_id,
"BRANCH_B_ID": branch_b_id,
"CLIENT_A_ID": client_a_id,
"CLIENT_B_ID": client_b_id,
"ENGAGEMENT_A_ID": v(sub_active, "id"),
"ENGAGEMENT_B_ID": v(sub_branch_b, "id"),
"SUBSCRIPTION_A_ID": v(sub_active, "id"),
"ACTIVE_FY_TASK_ID": v(task_active, "id"),
"TASK_A_ID": v(task_active, "id"),
"LOCKED_FY_TASK_ID": v(task_locked, "id"),
"PARTNER_TASK_A_ID": v(task_partner, "id"),
"BRANCH_B_TASK_ID": v(task_branch_b, "id"),
"TASK_DOCUMENT_A_ID": v(eng_doc, "id"),
"ENGAGEMENT_DOCUMENT_A_ID": v(eng_doc, "id"),
"PERM_DOCUMENT_ID": v(perm_doc, "id"),
"NOTICE_CASE_A_ID": v(notice_case, "id"),
"CASE_DOCUMENT_A_ID": v(notice_doc, "id"),
"INVOICE_A_ID": v(invoice, "id"),
"PAYMENT_A_ID": v(payment, "id"),
"ACTIVE_FY": ACTIVE_FY,
"LOCKED_FY": LOCKED_FY,
"DEFAULT_YEAR_CODE": ACTIVE_FY,
"WORK_TRACKER_ROUTE": "/services/work-tracker",
"WORK_TRACKER_FALLBACK_ROUTE": "/employee/work",
"DOCUMENTS_ROUTE": "/documents",
"PERMANENT_DOCUMENTS_ROUTE": "/documents/permanent",
"NOTICE_CASES_ROUTE": "/notice-cases",
"BILLING_ROUTE": "/billing",
"CLIENT_BILLING_ROUTE": "/client/billing",
}
print("\n[fy-strict-seed] Completed strict FY lock/backup seed patch.")
print("\nCopy/update these in /tests/.env if different:\n")
for k, val in env_ids.items():
print(f"{k}={val or ''}")
if notes:
print("\nNotes / non-fatal skips:")
for note in notes:
print(f"- {note}")
else:
print("\nNotes: no non-fatal seed skips reported.")
db.close()
if __name__ == "__main__":
main()