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