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 config.config import Config class TestConfig(Config): SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:' TESTING = True WTF_CSRF_ENABLED = False class Sprint4OptimizerAndConcurrencyTestCase(unittest.TestCase): """Pruebas exhaustivas para el Sprint 4: Optimizador Heurístico, Concurrencia y Spring Boot.""" def setUp(self): self.app = create_app(TestConfig) self.app_context = self.app.app_context() self.app_context.push() db.create_all() 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='ADMIN', 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( code=f"AULA-{100 + i}", building="Edificio Central", capacity=30 + (i * 5), 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.remove() db.drop_all() 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()