From 2d06a351c9da0a337bb4bed3b2838d9b46deb35c Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Mon, 22 Jun 2026 13:17:28 +0530 Subject: [PATCH] Add UAT seed data script --- scripts/seed_uat_data.py | 206 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 206 insertions(+) create mode 100644 scripts/seed_uat_data.py diff --git a/scripts/seed_uat_data.py b/scripts/seed_uat_data.py new file mode 100644 index 0000000..1f054f1 --- /dev/null +++ b/scripts/seed_uat_data.py @@ -0,0 +1,206 @@ +""" +Audit Firm ERP (Production_gitea) -- Playwright UAT/VAPT seed helper. + +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 + +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 + +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. +""" +from __future__ import annotations + +import os +import sys +from datetime import date, timedelta + +PASSWORD = os.getenv("UAT_SEED_PASSWORD", "Password@123") + +# Map UAT logical role -> actual Role.name in Production_gitea +ROLE_NAME = { + "firm_admin": "Firm Admin", + "partner": "Partner", + "manager": "Branch Manager", + "staff": "Staff", + "client": "Client", + "consultant": "Consultant", +} + + +def fail(msg: str) -> None: + print(f"\n[seed] ERROR: {msg}", file=sys.stderr) + sys.exit(1) + + +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.rbac.models import Role, UserRole + from app.modules.clients.models import Client + except Exception as exc: # pragma: no cover + fail( + "Could not import ERP modules. Run this from the ERP project root " + f"inside the app container. Underlying import error: {exc!r}" + ) + + # NoticeCase is optional (only if module present) + try: + from app.modules.notice_cases.models import NoticeCase + except Exception: + NoticeCase = None + + db = CommonSessionLocal() + ids: dict[str, object] = {} + + def get_or_create(Model, lookup: dict, defaults: dict): + obj = db.query(Model).filter_by(**lookup).first() + if obj: + 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}) + db.add(obj) + db.flush() + return obj, True + + def assign_role(user, role_label: str) -> None: + role_name = ROLE_NAME[role_label] + role = db.query(Role).filter(Role.name == role_name).first() + if not role: + print(f"[seed] WARNING: role '{role_name}' not found; " + f"run the app once so DEFAULT_ROLES are created. Skipping.") + return + exists = ( + db.query(UserRole) + .filter(UserRole.user_id == user.id, UserRole.role_id == role.id) + .first() + ) + if not exists: + db.add(UserRole(user_id=user.id, role_id=role.id)) + + 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}, + ) + 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) + + # --- Clients (note: client_name is required) -------------------------- + client_a, _ = get_or_create( + Client, {"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, + }, + ) + client_b, _ = get_or_create( + Client, {"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, + }, + ) + ids["CLIENT_A_ID"] = client_a.id + ids["CLIENT_B_ID"] = client_b.id + + # --- Notice case (optional) ------------------------------------------- + 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 + + 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}") + + except Exception: + db.rollback() + raise + finally: + db.close() + + +if __name__ == "__main__": + main()