mirror of
https://gitlab.com/megazordpobeda/DataRush.git
synced 2026-05-23 02:47:10 +00:00
feat: added dumb review, formatted
This commit is contained in:
@@ -0,0 +1,136 @@
|
||||
import random
|
||||
import uuid
|
||||
from datetime import timedelta
|
||||
|
||||
from django.contrib.auth.hashers import make_password
|
||||
from django.core.files.base import ContentFile
|
||||
from django.core.management.base import BaseCommand
|
||||
from django.utils import timezone
|
||||
|
||||
from apps.competition.models import Competition, State
|
||||
from apps.task.models import CompetetionTaskSumbission, CompetitionTask
|
||||
from apps.user.models import User, UserRole
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Generate sample data for Users, Competitions, Tasks, Submissions, and States."
|
||||
|
||||
def handle(self, *args, **options):
|
||||
self.stdout.write("Starting data generation...")
|
||||
users = self.create_users(5)
|
||||
competitions = self.create_competitions(2, users)
|
||||
tasks = self.create_tasks(competitions)
|
||||
self.create_submissions(tasks, users)
|
||||
self.create_states(competitions, users)
|
||||
self.stdout.write("Data generation completed.")
|
||||
|
||||
def create_users(self, count):
|
||||
users = []
|
||||
for i in range(1, count + 1):
|
||||
email = f"user{i}@example.com"
|
||||
username = f"user{i}"
|
||||
password = (
|
||||
"password123" # In production, use proper password handling.
|
||||
)
|
||||
role = random.choice(
|
||||
[UserRole.STUDENT.value, UserRole.METODIST.value]
|
||||
)
|
||||
user, created = User.objects.get_or_create(
|
||||
email=email,
|
||||
defaults={
|
||||
"username": username,
|
||||
"password": make_password(password),
|
||||
"status": role,
|
||||
},
|
||||
)
|
||||
users.append(user)
|
||||
self.stdout.write(f"Created user: {username}")
|
||||
return users
|
||||
|
||||
def create_competitions(self, count, users):
|
||||
competitions = []
|
||||
now = timezone.now()
|
||||
for i in range(1, count + 1):
|
||||
title = f"Competition {i}"
|
||||
description = f"Description for competition {i}"
|
||||
start_date = now - timedelta(days=random.randint(1, 10))
|
||||
end_date = now + timedelta(days=random.randint(1, 10))
|
||||
competition = Competition.objects.create(
|
||||
title=title,
|
||||
description=description,
|
||||
start_date=start_date,
|
||||
end_date=end_date,
|
||||
type="solo", # assuming only one type for now
|
||||
participation_type=random.choice(["edu", "competitive"]),
|
||||
)
|
||||
# Add random participants
|
||||
selected_users = random.sample(
|
||||
users, k=min(len(users), random.randint(1, len(users)))
|
||||
)
|
||||
competition.participants.add(*selected_users)
|
||||
competitions.append(competition)
|
||||
self.stdout.write(f"Created competition: {title}")
|
||||
return competitions
|
||||
|
||||
def create_tasks(self, competitions):
|
||||
tasks = []
|
||||
task_types = [
|
||||
CompetitionTask.CompetitionTaskType.INPUT.value,
|
||||
]
|
||||
for comp in competitions:
|
||||
# Create 3 tasks per competition
|
||||
for i in range(1, 4):
|
||||
task_type = random.choice(task_types)
|
||||
title = f"Task {i} for {comp.title}"
|
||||
description = f"Task description for task {i} in {comp.title}"
|
||||
task = CompetitionTask.objects.create(
|
||||
competition=comp,
|
||||
title=title,
|
||||
description=description,
|
||||
type=task_type,
|
||||
points=random.randint(1, 10),
|
||||
)
|
||||
tasks.append(task)
|
||||
self.stdout.write(f"Created task: {title} (type: {task_type})")
|
||||
return tasks
|
||||
|
||||
def create_submissions(self, tasks, users):
|
||||
for task in tasks:
|
||||
# Each task will get between 1 and 3 submissions
|
||||
num_submissions = random.randint(1, 3)
|
||||
for _ in range(num_submissions):
|
||||
user = random.choice(users)
|
||||
# Create a dummy content file
|
||||
dummy_content = ContentFile(
|
||||
b"Submission content",
|
||||
name=f"submission_{uuid.uuid4().hex}.txt",
|
||||
)
|
||||
submission = CompetetionTaskSumbission.objects.create(
|
||||
user=user,
|
||||
task=task,
|
||||
earned_points=random.randint(
|
||||
0, task.points if task.points else 10
|
||||
),
|
||||
content=dummy_content,
|
||||
)
|
||||
submission.save()
|
||||
self.stdout.write(
|
||||
f"Created submission for task '{task.title}' by user '{user.username}'"
|
||||
)
|
||||
|
||||
def create_states(self, competitions, users):
|
||||
# For each competition, create a State for some of its participants
|
||||
state_choices = [choice for choice in State.StateChoices.values]
|
||||
for comp in competitions:
|
||||
for user in comp.participants.all():
|
||||
state_obj, created = State.objects.get_or_create(
|
||||
user=user,
|
||||
competition=comp,
|
||||
defaults={
|
||||
"state": random.choice(state_choices),
|
||||
"changed_at": timezone.now(),
|
||||
},
|
||||
)
|
||||
self.stdout.write(
|
||||
f"Created state '{state_obj.state}' for user '{user.username}' in competition '{comp.title}'"
|
||||
)
|
||||
@@ -7,4 +7,4 @@ class Reviewer(BaseModel):
|
||||
name = models.CharField(max_length=100)
|
||||
surname = models.CharField(max_length=100)
|
||||
|
||||
token = models.CharField(max_length=100)
|
||||
token = models.CharField(max_length=100)
|
||||
|
||||
@@ -1,13 +1,13 @@
|
||||
from random import choice
|
||||
from uuid import uuid4
|
||||
|
||||
from django.db import models
|
||||
|
||||
from apps.task.validators import ContestTaskCriteriesValidator
|
||||
from apps.competition.models import Competition
|
||||
from apps.core.models import BaseModel
|
||||
from apps.task.validators import ContestTaskCriteriesValidator
|
||||
from apps.user.models import User
|
||||
|
||||
|
||||
class CompetitionTask(BaseModel):
|
||||
class CompetitionTaskType(models.TextChoices):
|
||||
INPUT = "input"
|
||||
@@ -45,7 +45,7 @@ class CompetetionTaskSumbission(BaseModel):
|
||||
CHECKED = "checked"
|
||||
|
||||
def submission_content_upload_to(instance, filename) -> str:
|
||||
return f"/submissions/{instance.id}/content"
|
||||
return f"submissions/{instance.id}/content"
|
||||
|
||||
def submission_stdout_upload_to(instance, filename) -> str:
|
||||
return f"/submissions/{instance.id}/stdout"
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import tempfile
|
||||
import ast
|
||||
import hashlib
|
||||
import os
|
||||
import sys
|
||||
import ast
|
||||
import tempfile
|
||||
from io import StringIO
|
||||
import hashlib
|
||||
|
||||
from config.celery import app
|
||||
|
||||
ALLOWED_MODULES = {
|
||||
@@ -29,7 +30,7 @@ def validate_code(code_str):
|
||||
try:
|
||||
tree = ast.parse(code_str)
|
||||
except SyntaxError as e:
|
||||
raise SecurityException(f"Syntax error: {str(e)}")
|
||||
raise SecurityException(f"Syntax error: {e!s}")
|
||||
|
||||
class ImportVisitor(ast.NodeVisitor):
|
||||
def visit_Import(self, node):
|
||||
@@ -56,10 +57,10 @@ def validate_code(code_str):
|
||||
try:
|
||||
ImportVisitor().visit(tree)
|
||||
SecurityVisitor().visit(tree)
|
||||
except SecurityException as e:
|
||||
except SecurityException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise SecurityException(f"Security check failed: {str(e)}")
|
||||
raise SecurityException(f"Security check failed: {e!s}")
|
||||
|
||||
|
||||
def secure_exec(code_str, result_path):
|
||||
@@ -95,7 +96,7 @@ def secure_exec(code_str, result_path):
|
||||
result_content = f.read()
|
||||
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Execution error: {str(e)}")
|
||||
raise RuntimeError(f"Execution error: {e!s}")
|
||||
finally:
|
||||
os.chdir(original_dir)
|
||||
sys.stdout = original_stdout
|
||||
@@ -121,8 +122,8 @@ def analyze_data_task(self, code_str, result_path, expected_bytes):
|
||||
}
|
||||
|
||||
except SecurityException as e:
|
||||
return {"success": False, "error": f"Security violation: {str(e)}"}
|
||||
return {"success": False, "error": f"Security violation: {e!s}"}
|
||||
except RuntimeError as e:
|
||||
return {"success": False, "error": f"Execution error: {str(e)}"}
|
||||
return {"success": False, "error": f"Execution error: {e!s}"}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": f"Unexpected error: {str(e)}"}
|
||||
return {"success": False, "error": f"Unexpected error: {e!s}"}
|
||||
|
||||
@@ -12,12 +12,12 @@ class Criteria(BaseModel):
|
||||
|
||||
class ContestTaskCriteriesValidator:
|
||||
def __call__(self, instance):
|
||||
if instance.criterties and not isinstance(instance.criterties, list):
|
||||
if instance.criteries and not isinstance(instance.criteries, list):
|
||||
err = "criteries must be a valid dictionary"
|
||||
raise ValidationError(err)
|
||||
|
||||
try:
|
||||
for criteria in instance.criterties:
|
||||
for criteria in instance.criteries if instance.criteries else []:
|
||||
Criteria(**criteria)
|
||||
except PydanticValidationError:
|
||||
err = "invalid criteries data"
|
||||
|
||||
@@ -24,9 +24,6 @@ class TestSignUp(TestCase):
|
||||
user.full_clean()
|
||||
|
||||
def test_missing_params(self):
|
||||
user = User(
|
||||
password="123123",
|
||||
username="132131232131"
|
||||
)
|
||||
user = User(password="123123", username="132131232131")
|
||||
with self.assertRaises(ValidationError):
|
||||
user.full_clean()
|
||||
|
||||
Reference in New Issue
Block a user