Initial Playwright ERP UAT VAPT test suite v2.4.1 IMAP

This commit is contained in:
A R R R Associates
2026-06-22 12:52:00 +05:30
parent 970d5df926
commit 865f19cb84
29 changed files with 26786 additions and 0 deletions
+225
View File
@@ -0,0 +1,225 @@
"""
Audit Firm ERP Playwright seed data helper (v2.2).
Run from your ERP project root after migrations:
python path\to\seed\seed_data.py
This script is intentionally defensive because different project phases may have
slightly different model names/columns. It uses SQLAlchemy model introspection,
creates only records where the model/table exists, and prints created IDs that
can be pasted into .env for cross-tenant/IDOR tests.
"""
from __future__ import annotations
import importlib
import os
import sys
from datetime import date, datetime, timedelta
from typing import Any, Dict, Optional
PASSWORD = os.getenv("UAT_SEED_PASSWORD", "Password@123")
def import_any(*paths):
for p in paths:
mod_name, attr = p.rsplit('.', 1)
try:
mod = importlib.import_module(mod_name)
return getattr(mod, attr)
except Exception:
continue
return None
def get_session():
SessionLocal = import_any(
'app.core.database.SessionLocal',
'app.core.db.SessionLocal',
'app.db.session.SessionLocal',
'app.database.SessionLocal',
)
if SessionLocal:
return SessionLocal()
raise RuntimeError('Could not locate SQLAlchemy SessionLocal. Run this from ERP project root or update import paths in seed_data.py.')
def get_model(name: str):
candidates = [
f'app.modules.core.iam.models.{name}',
f'app.modules.core.models.{name}',
f'app.modules.tenants.models.{name}',
f'app.modules.clients.models.{name}',
f'app.modules.services.models.{name}',
f'app.modules.notice_cases.models.{name}',
f'app.modules.employees.models.{name}',
f'app.modules.consultants.models.{name}',
f'app.models.{name}',
]
return import_any(*candidates)
def cols(Model):
return {c.name for c in Model.__table__.columns}
def pick(Model, **values):
c = cols(Model)
return {k: v for k, v in values.items() if k in c}
def first_by(db, Model, **criteria):
c = cols(Model)
q = db.query(Model)
for k, v in criteria.items():
if k in c:
q = q.filter(getattr(Model, k) == v)
return q.first()
def get_or_create(db, Model, defaults: Optional[Dict[str, Any]] = None, **criteria):
defaults = defaults or {}
obj = first_by(db, Model, **criteria)
if obj:
return obj, False
data = pick(Model, **criteria, **defaults)
obj = Model(**data)
db.add(obj)
db.flush()
return obj, True
def set_password(user):
# Try project hash helpers first.
helpers = [
'app.modules.core.iam.password_service.hash_password',
'app.modules.core.iam.security.hash_password',
'app.core.security.hash_password',
'app.core.auth.hash_password',
]
for h in helpers:
fn = import_any(h)
if fn:
for col in ('password_hash', 'hashed_password'):
if hasattr(user, col):
setattr(user, col, fn(PASSWORD))
return
# Fallback to passlib if available.
try:
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
hashed = pwd_context.hash(PASSWORD)
for col in ('password_hash', 'hashed_password'):
if hasattr(user, col):
setattr(user, col, hashed)
return
except Exception:
pass
def main():
db = get_session()
User = get_model('User')
Tenant = get_model('Tenant')
Branch = get_model('Branch')
Client = get_model('Client')
Service = get_model('Service')
Engagement = get_model('Engagement')
NoticeCase = get_model('NoticeCase')
ids: Dict[str, Any] = {}
try:
tenant_a = tenant_b = None
if Tenant:
tenant_a, _ = get_or_create(db, Tenant, name='UAT Tenant A', defaults={'code':'UAT-A','is_active':True}, code='UAT-A')
tenant_b, _ = get_or_create(db, Tenant, name='UAT Tenant B', defaults={'code':'UAT-B','is_active':True}, code='UAT-B')
ids['TENANT_A_ID'] = getattr(tenant_a, 'id', '')
ids['TENANT_B_ID'] = getattr(tenant_b, 'id', '')
branch_a = branch_b = None
if Branch:
branch_a, _ = get_or_create(db, Branch, name='UAT Branch A', defaults={'code':'UAT-BA','tenant_id':getattr(tenant_a,'id',None),'is_active':True}, code='UAT-BA')
branch_b, _ = get_or_create(db, Branch, name='UAT Branch B', defaults={'code':'UAT-BB','tenant_id':getattr(tenant_b,'id',None),'is_active':True}, code='UAT-BB')
ids['BRANCH_A_ID'] = getattr(branch_a, 'id', '')
ids['BRANCH_B_ID'] = getattr(branch_b, 'id', '')
if User:
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, tenant, branch in users:
criteria = {'email': email}
if 'login_id' in cols(User): criteria = {'login_id': email}
user, created = get_or_create(db, User, defaults={
'email': email, 'login_id': email, 'full_name': full_name, 'name': full_name,
'role': role, 'is_active': True, 'is_verified': True,
'tenant_id': getattr(tenant,'id',None), 'branch_id': getattr(branch,'id',None),
'must_change_password': False,
}, **criteria)
set_password(user)
client_a = client_b = None
if Client:
client_a, _ = get_or_create(db, Client, defaults={
'tenant_id':getattr(tenant_a,'id',None),'branch_id':getattr(branch_a,'id',None),
'client_code':'UAT-CL-A','name':'UAT Client A Pvt Ltd','company_name':'UAT Client A Pvt Ltd',
'pan':'AABCU1111A','gstin':'33AABCU1111A1Z5','email':'client.a@uat.test','mobile':'9000000001',
'status':'active','is_active':True
}, client_code='UAT-CL-A')
client_b, _ = get_or_create(db, Client, defaults={
'tenant_id':getattr(tenant_b,'id',None),'branch_id':getattr(branch_b,'id',None),
'client_code':'UAT-CL-B','name':'UAT Client B Pvt Ltd','company_name':'UAT Client B Pvt Ltd',
'pan':'AABCU2222A','gstin':'33AABCU2222A1Z5','email':'client.b@uat.test','mobile':'9000000002',
'status':'active','is_active':True
}, client_code='UAT-CL-B')
ids['CLIENT_A_ID'] = getattr(client_a, 'id', '')
ids['CLIENT_B_ID'] = getattr(client_b, 'id', '')
service_a = None
if Service:
service_a, _ = get_or_create(db, Service, defaults={
'tenant_id':getattr(tenant_a,'id',None),'branch_id':getattr(branch_a,'id',None),
'service_code':'UAT-SVC-GST','name':'UAT GST Compliance','description':'Seed service for Playwright tests',
'is_active':True,'base_rate':1000
}, service_code='UAT-SVC-GST')
ids['SERVICE_A_ID'] = getattr(service_a, 'id', '')
engagement_a = None
if Engagement:
engagement_a, _ = get_or_create(db, Engagement, defaults={
'tenant_id':getattr(tenant_a,'id',None),'branch_id':getattr(branch_a,'id',None),
'client_id':getattr(client_a,'id',None),'service_id':getattr(service_a,'id',None),
'engagement_code':'UAT-ENG-GST-001','title':'UAT GST Compliance Engagement',
'status':'open','financial_year':'2025-26','period':'2025-26','start_date':date.today()
}, engagement_code='UAT-ENG-GST-001')
ids['ENGAGEMENT_A_ID'] = getattr(engagement_a, 'id', '')
if NoticeCase:
nc, _ = get_or_create(db, NoticeCase, defaults={
'tenant_id':getattr(tenant_a,'id',None),'branch_id':getattr(branch_a,'id',None),
'client_id':getattr(client_a,'id',None),'department':'GST','case_type':'Notice',
'reference_no':'UAT-NOTICE-001','title':'UAT GST Notice','status':'Open',
'notice_date':date.today(),'due_date':date.today()+timedelta(days=15),
'description':'Seed notice case for Playwright tests'
}, reference_no='UAT-NOTICE-001')
ids['NOTICE_CASE_A_ID'] = getattr(nc, 'id', '')
db.commit()
print('\nSeed completed. Paste these values into the Playwright .env file if shown:')
for k, v in ids.items():
print(f'{k}={v}')
print('\nSeed user password for all uat.* users:', PASSWORD)
except Exception as exc:
db.rollback()
raise
finally:
db.close()
if __name__ == '__main__':
main()
+206
View File
@@ -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()