Backend Draft
This commit is contained in:
__init__
Binary file not shown.
294
backend/tenants/management/commands/populate_test_data.py
Normal file
294
backend/tenants/management/commands/populate_test_data.py
Normal file
@@ -0,0 +1,294 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user