276 lines
10 KiB
Python
276 lines
10 KiB
Python
import unittest
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
|
|
from app import create_app, db
|
|
from app.models.user import User
|
|
from app.models.role import Role
|
|
from app.models.classroom import Classroom
|
|
from app.models.genetic_algorithm import (
|
|
GeneticAlgorithm,
|
|
ReservationRequest,
|
|
Individual,
|
|
Gene
|
|
)
|
|
from app.services.optimizer_service import OptimizerService, OptimizerJobStatus
|
|
from app.services.jwt_service import JWTService
|
|
|
|
|
|
class Sprint4OptimizerAndConcurrencyTestCase(unittest.TestCase):
|
|
"""Pruebas exhaustivas para el Sprint 4: Optimizador Heurístico, Concurrencia y Spring Boot."""
|
|
|
|
def setUp(self):
|
|
self.app = create_app()
|
|
self.app.config['TESTING'] = True
|
|
self.app.config['WTF_CSRF_ENABLED'] = False
|
|
self.app_context = self.app.app_context()
|
|
self.app_context.push()
|
|
self.client = self.app.test_client()
|
|
|
|
# Obtener usuario existente o crear
|
|
self.user = User.query.filter_by(is_active=True).first()
|
|
if not self.user:
|
|
admin_role = Role.query.filter_by(name='Admin').first()
|
|
if not admin_role:
|
|
admin_role = Role(name='Admin', description='Administrador')
|
|
db.session.add(admin_role)
|
|
db.session.commit()
|
|
self.user = User(
|
|
name='Sprint 4 Admin',
|
|
email='admin_sprint4@edu-space.com',
|
|
role_id=admin_role.id,
|
|
is_active=True
|
|
)
|
|
self.user.set_password('admin123')
|
|
db.session.add(self.user)
|
|
db.session.commit()
|
|
|
|
# Obtener aulas activas de la base de datos
|
|
self.classrooms = Classroom.query.filter_by(is_active=True).all()
|
|
if not self.classrooms:
|
|
for i in range(1, 11):
|
|
room = Classroom(
|
|
room_number=f"AULA-{100 + i}",
|
|
capacity=30 + (i * 5),
|
|
is_virtual=(i == 10),
|
|
is_active=True
|
|
)
|
|
db.session.add(room)
|
|
db.session.commit()
|
|
self.classrooms = Classroom.query.filter_by(is_active=True).all()
|
|
|
|
# Token JWT para el usuario de prueba
|
|
token_pair = JWTService.generate_tokens(self.user)
|
|
self.access_token = token_pair['access_token']
|
|
self.auth_headers = {
|
|
'Authorization': f'Bearer {self.access_token}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
def tearDown(self):
|
|
db.session.rollback()
|
|
self.app_context.pop()
|
|
|
|
def test_01_fast_fitness_and_conflict_detection(self):
|
|
"""Verifica la detección de solapamiento O(K log K) y cálculo de fitness."""
|
|
now = datetime(2026, 3, 10, 18, 0, 0)
|
|
c1 = self.classrooms[0]
|
|
|
|
# Gene 1: 18:00 a 20:00 en AULA-101
|
|
g1 = Gene(
|
|
commission_id=1,
|
|
classroom_id=c1.id,
|
|
start_time=now,
|
|
end_time=now + timedelta(hours=2),
|
|
expected_attendees=30,
|
|
purpose="Clase Algoritmos"
|
|
)
|
|
# Gene 2: 19:00 a 21:00 en la MISMA aula (conflicto directo de 19:00 a 20:00)
|
|
g2 = Gene(
|
|
commission_id=2,
|
|
classroom_id=c1.id,
|
|
start_time=now + timedelta(hours=1),
|
|
end_time=now + timedelta(hours=3),
|
|
expected_attendees=30,
|
|
purpose="Clase Base de Datos"
|
|
)
|
|
|
|
ind = Individual([g1, g2])
|
|
classrooms_dict = {c.id: c for c in self.classrooms}
|
|
req1 = ReservationRequest(1, 30, "Clase Algoritmos", now, now + timedelta(hours=2))
|
|
req2 = ReservationRequest(2, 30, "Clase Base de Datos", now + timedelta(hours=1), now + timedelta(hours=3))
|
|
requests_dict = {1: req1, 2: req2}
|
|
|
|
fit = ind.calculate_fitness(classrooms_dict, requests_dict)
|
|
self.assertGreater(len(ind.conflicts), 0, "Debe detectar conflicto de solapamiento de horario")
|
|
|
|
def test_02_benchmark_200_commissions_under_5_seconds(self):
|
|
"""Hito de Evaluación Sprint 4: Optimización de 200+ comisiones en menos de 5 segundos."""
|
|
base_time = datetime(2026, 3, 16, 8, 0, 0)
|
|
requests = []
|
|
|
|
# Crear 200 peticiones de comisión distribuidas en turnos
|
|
for i in range(1, 201):
|
|
day_offset = (i % 5) # Lun a Vie
|
|
slot_offset = (i % 4) * 3 # 08:00, 11:00, 14:00, 18:00
|
|
start = base_time + timedelta(days=day_offset, hours=slot_offset)
|
|
req = ReservationRequest(
|
|
commission_id=i,
|
|
expected_attendees=25 + (i % 25),
|
|
purpose=f"Comisión {i}",
|
|
preferred_start_time=start,
|
|
preferred_end_time=start + timedelta(hours=2),
|
|
priority=1,
|
|
flexibility_hours=2
|
|
)
|
|
requests.append(req)
|
|
|
|
# Configurar optimizador con paralelización (4 workers)
|
|
ga = GeneticAlgorithm(
|
|
population_size=30,
|
|
generations=25,
|
|
workers=4
|
|
)
|
|
|
|
t_start = time.perf_counter()
|
|
best = ga.optimize_reservations(
|
|
requests=requests,
|
|
available_classrooms=self.classrooms,
|
|
start_date=base_time,
|
|
end_date=base_time + timedelta(days=7)
|
|
)
|
|
t_elapsed = time.perf_counter() - t_start
|
|
|
|
self.assertIsNotNone(best)
|
|
self.assertEqual(len(best.genes), 200, "Debe haber asignado las 200 comisiones")
|
|
self.assertLess(t_elapsed, 5.0, f"El algoritmo debe resolver 200 comisiones en < 5s (demoró {t_elapsed:.2f}s)")
|
|
|
|
def test_03_optimizer_service_async_queue(self):
|
|
"""Verifica la cola de trabajos asíncrona de OptimizerService."""
|
|
# Enviar un job en segundo plano con app_context
|
|
job = OptimizerService.submit_job(
|
|
commission_ids=[1, 2],
|
|
admin_user_id=self.user.id,
|
|
generations=10,
|
|
population_size=10,
|
|
workers=2,
|
|
app=self.app
|
|
)
|
|
|
|
self.assertIsNotNone(job.job_id)
|
|
self.assertIn(job.status, [OptimizerJobStatus.PENDING, OptimizerJobStatus.RUNNING, OptimizerJobStatus.COMPLETED])
|
|
|
|
# Esperar a que el worker complete la ejecución (máximo 4 segundos)
|
|
waited = 0.0
|
|
while job.status in (OptimizerJobStatus.PENDING, OptimizerJobStatus.RUNNING) and waited < 4.0:
|
|
time.sleep(0.1)
|
|
waited += 0.1
|
|
|
|
self.assertEqual(job.status, OptimizerJobStatus.COMPLETED, f"El trabajo debió completar con éxito, error: {job.error}")
|
|
self.assertEqual(job.progress, 100)
|
|
self.assertIsNotNone(job.completed_at)
|
|
self.assertIsNotNone(job.execution_time_ms)
|
|
|
|
# Consultar via list_jobs y get_job
|
|
found = OptimizerService.get_job(job.job_id)
|
|
self.assertIsNotNone(found)
|
|
self.assertEqual(found.job_id, job.job_id)
|
|
|
|
all_jobs = OptimizerService.list_jobs()
|
|
self.assertTrue(any(j['job_id'] == job.job_id for j in all_jobs))
|
|
|
|
def test_04_spring_boot_sync_payload_endpoint(self):
|
|
"""Verifica el endpoint de interoperabilidad REST/JWT con Spring Boot."""
|
|
now = datetime(2026, 4, 1, 18, 30, 0)
|
|
payload = {
|
|
"requestId": "REQ-SPRING-BOOT-2026-001",
|
|
"campusCode": "SEDE-CENTRAL",
|
|
"academicTerm": "2026-1C",
|
|
"commissions": [
|
|
{
|
|
"commissionId": 101,
|
|
"subjectCode": "INFO-101",
|
|
"subjectName": "Programación Orientada a Objetos",
|
|
"expectedAttendees": 35,
|
|
"preferredShift": "NOCHE",
|
|
"preferredStart": now.isoformat(),
|
|
"preferredEnd": (now + timedelta(hours=3)).isoformat()
|
|
},
|
|
{
|
|
"commissionId": 102,
|
|
"subjectCode": "MATH-201",
|
|
"subjectName": "Análisis Matemático II",
|
|
"expectedAttendees": 40,
|
|
"preferredShift": "NOCHE",
|
|
"preferredStart": now.isoformat(),
|
|
"preferredEnd": (now + timedelta(hours=3)).isoformat()
|
|
}
|
|
],
|
|
"options": {
|
|
"generations": 15,
|
|
"populationSize": 15,
|
|
"workers": 2
|
|
}
|
|
}
|
|
|
|
response = self.client.post(
|
|
'/api/v1/optimizer/spring-boot/sync',
|
|
json=payload,
|
|
headers=self.auth_headers
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.get_json()
|
|
self.assertEqual(data.get('requestId'), "REQ-SPRING-BOOT-2026-001")
|
|
self.assertEqual(data.get('status'), "SUCCESS")
|
|
self.assertEqual(data.get('totalCommissions'), 2)
|
|
self.assertIn('scheduledAssignments', data)
|
|
self.assertIn('executionTimeMs', data)
|
|
self.assertEqual(len(data['scheduledAssignments']), 2)
|
|
|
|
def test_05_api_async_job_submission_and_polling(self):
|
|
"""Verifica la API REST de despacho asíncrono y sondeo (polling)."""
|
|
payload = {
|
|
"commission_ids": [1, 2],
|
|
"generations": 10,
|
|
"population_size=10": None, # será ignorado o testeado con defaults
|
|
"generations": 10,
|
|
"population_size": 10,
|
|
"workers": 2
|
|
}
|
|
|
|
# 1. Enviar job (debe retornar 202 Accepted)
|
|
post_res = self.client.post(
|
|
'/api/v1/optimizer/jobs',
|
|
json=payload,
|
|
headers=self.auth_headers
|
|
)
|
|
self.assertEqual(post_res.status_code, 202)
|
|
post_data = post_res.get_json()
|
|
self.assertIn('job_id', post_data)
|
|
job_id = post_data['job_id']
|
|
self.assertEqual(post_data['poll_url'], f'/api/v1/optimizer/jobs/{job_id}')
|
|
|
|
# 2. Consultar estado del job
|
|
get_res = self.client.get(
|
|
f'/api/v1/optimizer/jobs/{job_id}',
|
|
headers=self.auth_headers
|
|
)
|
|
self.assertEqual(get_res.status_code, 200)
|
|
get_data = get_res.get_json()
|
|
self.assertEqual(get_data['job_id'], job_id)
|
|
self.assertIn(get_data['status'], ['PENDING', 'RUNNING', 'COMPLETED'])
|
|
|
|
# 3. Listar trabajos
|
|
list_res = self.client.get(
|
|
'/api/v1/optimizer/jobs',
|
|
headers=self.auth_headers
|
|
)
|
|
self.assertEqual(list_res.status_code, 200)
|
|
list_data = list_res.get_json()
|
|
self.assertGreaterEqual(list_data['total'], 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|