""" 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()