295 lines
12 KiB
Python
295 lines
12 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Django Management Command: populate_test_data
|
|
|
|
Creates realistic test data for MTCBD.
|
|
Run with: python manage.py populate_test_data
|
|
"""
|
|
|
|
from django.core.management.base import BaseCommand
|
|
from faker import Faker
|
|
from django.utils import timezone
|
|
import random
|
|
import datetime
|
|
|
|
from tenants.models import Tenant
|
|
from accounts.models import User
|
|
from projects.models import Project, Task
|
|
from dashboard.models import Notification
|
|
from analytics.models import ActivityLog
|
|
|
|
fake = Faker()
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Populate database with realistic test data (100+ users, projects, tasks, etc.)'
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
'--clear',
|
|
action='store_true',
|
|
help='Clear existing data before populating',
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
clear = options['clear']
|
|
|
|
if clear and Tenant.objects.exists():
|
|
self.stdout.write(self.style.WARNING('Clearing existing data...'))
|
|
Task.objects.all().delete()
|
|
Project.objects.all().delete()
|
|
Notification.objects.all().delete()
|
|
ActivityLog.objects.all().delete()
|
|
User.objects.all().delete()
|
|
Tenant.objects.all().delete()
|
|
self.stdout.write(self.style.SUCCESS('✓ Data cleared'))
|
|
|
|
self.main()
|
|
self.stdout.write(self.style.SUCCESS('\n✅ Population complete!'))
|
|
|
|
def main(self):
|
|
self.stdout.write("MTCBD Database Population".center(60))
|
|
self.stdout.write("="*60)
|
|
|
|
# Configuration
|
|
NUM_TENANTS = 5
|
|
USERS_PER_TENANT = 20
|
|
PROJECTS_PER_TENANT = (3, 8)
|
|
TASKS_PER_PROJECT = (5, 15)
|
|
NOTIFICATIONS_PER_USER = (0, 5)
|
|
ACTIVITY_LOGS_PER_TENANT = 50
|
|
|
|
TENANT_DATA = [
|
|
{'name': 'Tech Academy', 'subdomain': 'techacademy'},
|
|
{'name': 'Medical College', 'subdomain': 'medcollege'},
|
|
{'name': 'Business School', 'subdomain': 'businessschool'},
|
|
{'name': 'Engineering Institute', 'subdomain': 'enginstitute'},
|
|
{'name': 'Arts University', 'subdomain': 'artsuniv'},
|
|
]
|
|
|
|
tenants = self.create_tenants(TENANT_DATA)
|
|
total_users = 0
|
|
total_projects = 0
|
|
total_tasks = 0
|
|
total_notifications = 0
|
|
total_logs = 0
|
|
|
|
for tenant in tenants:
|
|
users = self.create_users_for_tenant(tenant, USERS_PER_TENANT)
|
|
total_users += len(users)
|
|
|
|
projects = self.create_projects_for_tenant(tenant, users, PROJECTS_PER_TENANT)
|
|
total_projects += len(projects)
|
|
|
|
if projects:
|
|
tasks_created = self.create_tasks_for_projects(projects, users, TASKS_PER_PROJECT)
|
|
total_tasks += tasks_created
|
|
|
|
notifications = self.create_notifications(users, NOTIFICATIONS_PER_USER)
|
|
total_notifications += notifications
|
|
|
|
logs = self.create_activity_logs(tenant, users, ACTIVITY_LOGS_PER_TENANT)
|
|
total_logs += logs
|
|
|
|
self.print_summary(tenants, total_users, total_projects, total_tasks, total_notifications, total_logs)
|
|
|
|
def create_tenants(self, tenant_data):
|
|
self.stdout.write('\nCreating tenants...')
|
|
tenants = []
|
|
for data in tenant_data:
|
|
tenant, created = Tenant.objects.get_or_create(
|
|
subdomain=data['subdomain'],
|
|
defaults={'name': data['name'], 'is_active': True}
|
|
)
|
|
tenants.append(tenant)
|
|
status = "✓ Created" if created else "→ Exists"
|
|
self.stdout.write(f' {status}: {tenant.name} ({tenant.subdomain})')
|
|
return tenants
|
|
|
|
def create_users_for_tenant(self, tenant, count):
|
|
self.stdout.write(f'\n Creating users for {tenant.subdomain}...')
|
|
users = []
|
|
|
|
# One super_admin and one institution_admin
|
|
admin_roles = ['super_admin', 'institution_admin']
|
|
for role in admin_roles:
|
|
username = f"{role}_{tenant.subdomain}"
|
|
email = f"{role}@{tenant.subdomain}.com"
|
|
user, created = User.objects.get_or_create(
|
|
username=username,
|
|
defaults={
|
|
'email': email,
|
|
'first_name': fake.first_name(),
|
|
'last_name': fake.last_name(),
|
|
'role': role,
|
|
'tenant': tenant,
|
|
'is_staff': role in ['super_admin', 'institution_admin'],
|
|
'is_active': True,
|
|
}
|
|
)
|
|
if created:
|
|
user.set_password('password123')
|
|
user.save()
|
|
users.append(user)
|
|
self.stdout.write(f' ✓ {role}: {username}')
|
|
|
|
# Other users
|
|
remaining = count - 2
|
|
for i in range(remaining):
|
|
role = random.choice(['teacher', 'project_manager', 'student', 'student', 'student'])
|
|
username = f"{tenant.subdomain}_{i+1}"
|
|
email = f"user{i+1}@{tenant.subdomain}.com"
|
|
user, created = User.objects.get_or_create(
|
|
username=username,
|
|
defaults={
|
|
'email': email,
|
|
'first_name': fake.first_name(),
|
|
'last_name': fake.last_name(),
|
|
'role': role,
|
|
'tenant': tenant,
|
|
'is_active': True,
|
|
}
|
|
)
|
|
if created:
|
|
user.set_password('password123')
|
|
user.save()
|
|
users.append(user)
|
|
|
|
self.stdout.write(f' → Total users for {tenant.subdomain}: {len(users)}')
|
|
return users
|
|
|
|
def create_projects_for_tenant(self, tenant, users, range_tuple):
|
|
self.stdout.write(f'\n Creating projects for {tenant.subdomain}...')
|
|
projects = []
|
|
num_projects = random.randint(*range_tuple)
|
|
statuses = ['active', 'planned', 'completed', 'archived']
|
|
|
|
admin_users = [u for u in users if u.role in ['teacher', 'institution_admin', 'super_admin', 'project_manager']]
|
|
if not admin_users:
|
|
admin_users = users
|
|
|
|
for i in range(num_projects):
|
|
creator = random.choice(admin_users)
|
|
project = Project.objects.create(
|
|
tenant=tenant,
|
|
name=fake.catch_phrase().title(),
|
|
description=fake.text(max_nb_chars=200) if random.random() > 0.3 else '',
|
|
created_by=creator,
|
|
status=random.choice(statuses),
|
|
created_at=random_date(timezone.now() - datetime.timedelta(days=365), timezone.now()),
|
|
)
|
|
projects.append(project)
|
|
self.stdout.write(f' ✓ Project: {project.name}')
|
|
return projects
|
|
|
|
def create_tasks_for_projects(self, projects, users, range_tuple):
|
|
self.stdout.write(f'\n Creating tasks...')
|
|
total_tasks = 0
|
|
statuses = ['todo', 'in_progress', 'review', 'done']
|
|
priorities = ['low', 'medium', 'high', 'urgent']
|
|
|
|
for project in projects:
|
|
num_tasks = random.randint(*range_tuple)
|
|
project_users = [u for u in users if u.tenant == project.tenant]
|
|
|
|
for i in range(num_tasks):
|
|
assigned_to = random.choice(project_users) if random.random() > 0.3 else None
|
|
due_date = random_date(timezone.now() + datetime.timedelta(days=1), timezone.now() + datetime.timedelta(days=90)) if random.random() > 0.4 else None
|
|
|
|
task = Task.objects.create(
|
|
tenant=project.tenant,
|
|
project=project,
|
|
title=fake.sentence(nb_words=6).rstrip('.'),
|
|
description=fake.text(max_nb_chars=150) if random.random() > 0.5 else '',
|
|
assigned_to=assigned_to,
|
|
status=random.choice(statuses),
|
|
priority=random.choice(priorities),
|
|
due_date=due_date,
|
|
)
|
|
total_tasks += 1
|
|
self.stdout.write(f' → {project.name}: {num_tasks} tasks')
|
|
|
|
self.stdout.write(f' ✓ Total tasks created: {total_tasks}')
|
|
return total_tasks
|
|
|
|
def create_notifications(self, users, range_tuple):
|
|
self.stdout.write(f'\n Creating notifications...')
|
|
total_notifications = 0
|
|
titles = [
|
|
'New task assigned', 'Project update', 'Deadline approaching',
|
|
'Comment on your task', 'Project completed', 'Meeting reminder',
|
|
]
|
|
|
|
for user in users:
|
|
num_notifications = random.randint(*range_tuple)
|
|
for i in range(num_notifications):
|
|
Notification.objects.create(
|
|
tenant=user.tenant,
|
|
user=user,
|
|
title=random.choice(titles),
|
|
message=fake.sentence(),
|
|
is_read=random.choice([True, False, False, False]),
|
|
link=fake.url() if random.random() > 0.7 else None,
|
|
created_at=random_date(timezone.now() - datetime.timedelta(days=30), timezone.now()),
|
|
)
|
|
total_notifications += 1
|
|
self.stdout.write(f' ✓ Total notifications: {total_notifications}')
|
|
return total_notifications
|
|
|
|
def create_activity_logs(self, tenant, users, count):
|
|
self.stdout.write(f'\n Creating activity logs for {tenant.subdomain}...')
|
|
actions = ['created', 'updated', 'deleted', 'logged_in', 'assigned', 'completed']
|
|
target_types = ['project', 'task', 'user', 'notification']
|
|
|
|
all_users = list(User.objects.all())
|
|
total_logs = 0
|
|
|
|
for i in range(count):
|
|
user = random.choice(all_users) if all_users else None
|
|
ActivityLog.objects.create(
|
|
tenant=tenant,
|
|
user=user,
|
|
action=random.choice(actions),
|
|
target_type=random.choice(target_types),
|
|
target_id=str(random.randint(1, 1000)),
|
|
metadata={
|
|
'ip_address': fake.ipv4(),
|
|
'user_agent': fake.user_agent(),
|
|
} if random.random() > 0.5 else {},
|
|
created_at=random_date(timezone.now() - datetime.timedelta(days=180), timezone.now()),
|
|
)
|
|
total_logs += 1
|
|
|
|
self.stdout.write(f' ✓ Activity logs: {total_logs}')
|
|
return total_logs
|
|
|
|
def print_summary(self, tenants, total_users, total_projects, total_tasks, total_notifications, total_logs):
|
|
self.stdout.write("\n" + "="*60)
|
|
self.stdout.write(" POPULATION COMPLETE".center(60))
|
|
self.stdout.write("="*60)
|
|
self.stdout.write(f' Tenants created: {len(tenants)}')
|
|
self.stdout.write(f' Total users: {total_users}')
|
|
self.stdout.write(f' Total projects: {total_projects}')
|
|
self.stdout.write(f' Total tasks: {total_tasks}')
|
|
self.stdout.write(f' Total notifications: {total_notifications}')
|
|
self.stdout.write(f' Total activity logs: {total_logs}')
|
|
self.stdout.write("="*60)
|
|
self.stdout.write('\nTest accounts per tenant:')
|
|
self.stdout.write(' Username: super_admin_<subdomain> / Password: password123')
|
|
self.stdout.write(' Username: institution_admin_<subdomain> / Password: password123')
|
|
self.stdout.write(' Other users: <subdomain>_1, <subdomain>_2, ...')
|
|
self.stdout.write('\nLogin example (use first tenant ID as X-Tenant-ID):')
|
|
self.stdout.write(' POST /api/auth/login/')
|
|
self.stdout.write(' Headers: X-Tenant-ID: 1')
|
|
self.stdout.write(' Body: {"username": "super_admin_techacademy", "password": "password123"}')
|
|
self.stdout.write("="*60)
|
|
|
|
# Also print tenant IDs for reference
|
|
self.stdout.write('\nTenant IDs:')
|
|
for t in tenants:
|
|
self.stdout.write(f' {t.id}: {t.name} ({t.subdomain})')
|
|
|
|
def random_date(start_date, end_date):
|
|
delta = end_date - start_date
|
|
random_days = random.randint(0, delta.days)
|
|
return start_date + datetime.timedelta(days=random_days)
|