carlostellocba/admin-edu-space
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: completar Fase 2 del Roadmap MVP (Co-docencia, Matriz de Conflictos y Regla Diaria) y auditoria de seguridad

Browse Source
This commit is contained in:
carlostellocba
2026-09-19 14:05:16 -03:00
parent 05e197e49f
commit ada39248d7
36 changed files with 3037 additions and 443 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/__init__.py
+2
View File
@@ -29,6 +29,7 @@ def create_app(config_class=Config):
from app.routes.api.optimizer import api_optimizer_bp
from app.routes.api.dashboard import api_dashboard_bp
from app.routes.api.admin import api_admin_bp
from app.routes.api.openapi import api_openapi_bp
from app.security import SecurityFilterChain
# Register API v1 blueprints
@@ -37,6 +38,7 @@ def create_app(config_class=Config):
app.register_blueprint(api_reservations_bp)
app.register_blueprint(api_optimizer_bp)
app.register_blueprint(api_dashboard_bp)
app.register_blueprint(api_openapi_bp)
app.register_blueprint(api_admin_bp, url_prefix='/api/v1')
app.register_blueprint(api_admin_bp, url_prefix='/api/v1/admin', name='api_admin_prefixed')
backend/app/constants/__init__.py
+1
View File
@@ -0,0 +1 @@
from .document_types import DOCUMENT_TYPES, DOCUMENT_TYPE_MAP, get_document_type, validate_document, format_document
backend/app/constants/document_types.py
+160
View File
@@ -0,0 +1,160 @@
import re
DOCUMENT_TYPES = [
{
'code': 'DNI',
'name': 'Documento Nacional de Identidad',
'short_name': 'DNI',
'category': 'Nacional',
'description': 'Documento oficial para ciudadanos argentinos nativos, naturalizados y residentes permanentes.',
'regex': r'^\d{7,8}$',
'placeholder': 'Ej: 38123456 (7 u 8 dígitos sin puntos)',
'mask_example': '38.123.456'
},
{
'code': 'CUIL',
'name': 'Código Único de Identificación Laboral (CUIL/CUIT)',
'short_name': 'CUIL',
'category': 'Nacional / Laboral',
'description': 'Identificador fiscal y laboral en Argentina (11 dígitos).',
'regex': r'^\d{11}$|^\d{2}-\d{8}-\d{1}$',
'placeholder': 'Ej: 20381234567 o 20-38123456-7',
'mask_example': '20-38123456-7'
},
{
'code': 'PAS',
'name': 'Pasaporte Argentino',
'short_name': 'Pasaporte ARG',
'category': 'Internacional',
'description': 'Pasaporte emitido por la República Argentina para viajes y acreditación internacional.',
'regex': r'^[a-zA-Z0-9]{6,12}$',
'placeholder': 'Ej: AAB123456',
'mask_example': 'PAS AAB123456'
},
{
'code': 'PASEXT',
'name': 'Pasaporte Extranjero',
'short_name': 'Pasaporte Extranjero',
'category': 'Internacional / Extranjero',
'description': 'Pasaporte oficial emitido por país de origen (docentes invitados, alumnos de intercambio).',
'regex': r'^[a-zA-Z0-9\-\.]{4,20}$',
'placeholder': 'Ej: US987654321 / B1234567',
'mask_example': 'PASEXT US987654321'
},
{
'code': 'DNIEXT',
'name': 'DNI para Extranjeros (Residente)',
'short_name': 'DNI Extranjero',
'category': 'Extranjero / Residente',
'description': 'DNI argentino otorgado a ciudadanos extranjeros con radicación o residencia temporaria.',
'regex': r'^[a-zA-Z0-9]{7,10}$',
'placeholder': 'Ej: 94123456 o E94123456',
'mask_example': 'DNI-EXT 94.123.456'
},
{
'code': 'CI',
'name': 'Cédula de Identidad (PFA / Policía Provincial)',
'short_name': 'Cédula Identidad',
'category': 'Nacional (Histórico)',
'description': 'Cédula de Identidad emitida por Policía Federal Argentina o Policías Provinciales.',
'regex': r'^\d{6,9}$',
'placeholder': 'Ej: 6123456',
'mask_example': 'CI 6.123.456'
},
{
'code': 'CIEXT',
'name': 'Cédula de Identidad Extranjera (Mercosur)',
'short_name': 'Cédula Mercosur',
'category': 'Regional / Mercosur',
'description': 'Cédula de Identidad emitida por países miembros o asociados al Mercosur (Brasil, Uruguay, etc.).',
'regex': r'^[a-zA-Z0-9\-\.]{5,18}$',
'placeholder': 'Ej: RG12345678 (Brasil) o 4.123.456-7 (Uruguay)',
'mask_example': 'CI-EXT 12345678'
},
{
'code': 'LC',
'name': 'Libreta Cívica',
'short_name': 'Libreta Cívica',
'category': 'Histórico Nacional',
'description': 'Documento histórico nacional para ciudadanas mujeres argentinas.',
'regex': r'^\d{6,8}$',
'placeholder': 'Ej: 5123456',
'mask_example': 'LC 5.123.456'
},
{
'code': 'LE',
'name': 'Libreta de Enrolamiento',
'short_name': 'Libreta Enrolamiento',
'category': 'Histórico Nacional',
'description': 'Documento histórico nacional para ciudadanos varones argentinos.',
'regex': r'^\d{6,8}$',
'placeholder': 'Ej: 4123456',
'mask_example': 'LE 4.123.456'
}
]
DOCUMENT_TYPE_MAP = {doc['code']: doc for doc in DOCUMENT_TYPES}
def get_document_type(code):
"""Devuelve la especificación de un tipo de documento o None."""
if not code:
return None
return DOCUMENT_TYPE_MAP.get(str(code).strip().upper())
def validate_document(doc_type, doc_number):
"""
Valida un número de documento según las reglas de negocio de su tipificación.
Retorna (is_valid: bool, cleaned_number: str, error_message: str|None).
"""
if not doc_type:
return False, '', 'Debe especificar el tipo de documento.'
spec = get_document_type(doc_type)
if not spec:
return False, '', f"El tipo de documento '{doc_type}' no está registrado en el catálogo oficial."
if not doc_number:
return True, '', None # Document number may be optional if pending registration
cleaned = str(doc_number).strip().upper()
# Si es DNI, LC, LE, CI podemos remover puntos y guiones intermedios para validar y almacenar de forma uniforme
if spec['code'] in ['DNI', 'LC', 'LE', 'CI', 'DNIEXT']:
numeric_only = re.sub(r'[\.\-\s]', '', cleaned)
if re.match(spec['regex'], numeric_only):
return True, numeric_only, None
return False, cleaned, f"Formato inválido para {spec['name']}. Debe contener {spec['placeholder']}."
elif spec['code'] == 'CUIL':
numeric_only = re.sub(r'[\.\-\s]', '', cleaned)
if re.match(r'^\d{11}$', numeric_only):
return True, numeric_only, None
return False, cleaned, "El CUIL debe contener 11 dígitos numéricos (ej. 20-38123456-7)."
else:
# Pasaportes o Cédulas Extranjeras
if re.match(spec['regex'], cleaned):
return True, cleaned, None
return False, cleaned, f"Formato inválido para {spec['name']} ({spec['placeholder']})."
def format_document(doc_type, doc_number):
"""
Formatea visualmente un número de documento con puntos/guiones estándar.
"""
if not doc_number:
return ''
raw = str(doc_number).strip().upper()
dtype = (doc_type or 'DNI').upper()
if dtype in ['DNI', 'LC', 'LE', 'CI']:
clean = re.sub(r'\D', '', raw)
if len(clean) == 8:
return f"{clean[:2]}.{clean[2:5]}.{clean[5:]}"
elif len(clean) == 7:
return f"{clean[:1]}.{clean[1:4]}.{clean[4:]}"
return clean
elif dtype == 'CUIL':
clean = re.sub(r'\D', '', raw)
if len(clean) == 11:
return f"{clean[:2]}-{clean[2:10]}-{clean[10:]}"
return clean
return raw
backend/app/models/__init__.py
+2 -1
View File
@@ -3,7 +3,7 @@ from .building import Building
from .classroom import Classroom, ClassroomResource
from .career import Career
from .academic_term import AcademicTerm
from .subject import Subject, Commission
from .subject import Subject, Commission, CommissionTeacher
from .reservation import Reservation, ReservationStatus
from .genetic_algorithm import GeneticAlgorithm, ReservationOptimizer
from .role import Role, Permission, SYSTEM_MODULES
@@ -21,6 +21,7 @@ __all__ = [
'AcademicTerm',
'Subject',
'Commission',
'CommissionTeacher',
'Reservation',
'ReservationStatus',
'GeneticAlgorithm',
backend/app/models/enrollment.py
+4
View File
@@ -23,6 +23,8 @@ class StudentEnrollment(db.Model):
enrolled_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
notes = db.Column(db.Text, nullable=True)
allow_same_day_exception = db.Column(db.Boolean, default=False, nullable=True)
exception_reason = db.Column(db.String(255), nullable=True)
# Unique constraint: a student can only be enrolled once per commission
__table_args__ = (
@@ -43,6 +45,8 @@ class StudentEnrollment(db.Model):
'status': self.status,
'enrolled_at': self.enrolled_at.isoformat() if self.enrolled_at else None,
'notes': self.notes,
'allow_same_day_exception': bool(self.allow_same_day_exception),
'exception_reason': self.exception_reason
}
def __repr__(self):
backend/app/models/subject.py
+185 -118
View File
@@ -1,118 +1,185 @@
from app import db
from datetime import datetime
class Subject(db.Model):
__tablename__ = 'subjects'
id = db.Column(db.Integer, primary_key=True)
code = db.Column(db.String(50), unique=True, nullable=False, index=True)
name = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
department = db.Column(db.String(100), nullable=False, default='')
credits = db.Column(db.Integer, nullable=False, default=4)
career_id = db.Column(db.Integer, db.ForeignKey('careers.id', ondelete='SET NULL'), nullable=True)
is_active = db.Column('active', db.Boolean, default=True, nullable=False) # Maps to active column in DB
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
commissions = db.relationship('Commission', backref='subject', lazy=True, cascade='all, delete-orphan')
def to_dict(self):
"""Convert subject to dictionary"""
return {
'id': self.id,
'code': self.code,
'name': self.name,
'description': self.description,
'department': self.department,
'credits': self.credits,
'career_id': self.career_id,
'career_name': self.career_obj.name if self.career_obj else None,
'is_active': self.is_active,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
def __repr__(self):
return f'<Subject {self.code}>'
class Commission(db.Model):
__tablename__ = 'commissions'
id = db.Column(db.Integer, primary_key=True)
subject_id = db.Column(db.Integer, db.ForeignKey('subjects.id'), nullable=False)
code = db.Column(db.String(50), nullable=False)
semester = db.Column(db.String(20), nullable=False)
year = db.Column(db.Integer, nullable=False)
teacher_id = db.Column(db.Integer, db.ForeignKey('users.id'))
max_students = db.Column(db.Integer, nullable=False)
current_students = db.Column(db.Integer, default=0)
schedule = db.Column(db.String(255))
shift = db.Column(db.String(30), nullable=True) # Mañana, Tarde, Vespertino, Noche
virtual_link = db.Column(db.String(500), nullable=True) # Link Zoom, Meet, Teams, Moodle
active = db.Column(db.Boolean, default=True) # Changed from is_active to active
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Virtual property for compatibility
@property
def is_active(self):
return self.active
@is_active.setter
def is_active(self, value):
self.active = value
# Relationships
pass # Reservation already defines the relationship back to commission
teacher = db.relationship('User', backref='teaching_commissions', foreign_keys=[teacher_id])
# Virtual field for teacher_name
@property
def teacher_name(self):
return f"{self.teacher.first_name} {self.teacher.last_name}" if self.teacher else "TBD"
# Unique constraint
__table_args__ = (db.UniqueConstraint('subject_id', 'code', 'semester', 'year'),)
def to_dict(self):
"""Convert commission to dictionary"""
return {
'id': self.id,
'subject_id': self.subject_id,
'code': self.code,
'semester': self.semester,
'year': self.year,
'teacher_name': self.teacher_name,
'max_students': self.max_students,
'current_students': self.current_students,
'schedule': self.schedule,
'shift': self.shift,
'virtual_link': self.virtual_link,
'active': self.active,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'full_code': self.get_full_code(),
'subject': self.subject.to_dict() if self.subject else None
}
def get_full_code(self):
"""Get full commission code"""
subj = self.subject.code if self.subject else 'N/A'
code = self.code or ''
sem = str(self.semester or '').strip()
yr = str(self.year or '').strip()
if yr and yr in sem:
return f"{subj}-{code}-{sem}"
elif sem and yr:
return f"{subj}-{code}-{sem}{yr}"
elif sem:
return f"{subj}-{code}-{sem}"
elif yr:
return f"{subj}-{code}-{yr}"
return f"{subj}-{code}"
def __repr__(self):
return f'<Commission {self.get_full_code()}>'
from app import db
from datetime import datetime
class Subject(db.Model):
__tablename__ = 'subjects'
id = db.Column(db.Integer, primary_key=True)
code = db.Column(db.String(50), unique=True, nullable=False, index=True)
name = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
department = db.Column(db.String(100), nullable=False, default='')
credits = db.Column(db.Integer, nullable=False, default=4)
career_id = db.Column(db.Integer, db.ForeignKey('careers.id', ondelete='SET NULL'), nullable=True)
is_active = db.Column('active', db.Boolean, default=True, nullable=False) # Maps to active column in DB
is_short_course = db.Column(db.Boolean, default=False, nullable=True) # True para talleres/cursos cortos
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
commissions = db.relationship('Commission', backref='subject', lazy=True, cascade='all, delete-orphan')
def to_dict(self):
"""Convert subject to dictionary"""
return {
'id': self.id,
'code': self.code,
'name': self.name,
'description': self.description,
'department': self.department,
'credits': self.credits,
'career_id': self.career_id,
'career_name': self.career_obj.name if self.career_obj else None,
'is_active': self.is_active,
'is_short_course': bool(self.is_short_course),
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
def __repr__(self):
return f'<Subject {self.code}>'
class Commission(db.Model):
__tablename__ = 'commissions'
id = db.Column(db.Integer, primary_key=True)
subject_id = db.Column(db.Integer, db.ForeignKey('subjects.id'), nullable=False)
code = db.Column(db.String(50), nullable=False)
semester = db.Column(db.String(20), nullable=False)
year = db.Column(db.Integer, nullable=False)
teacher_id = db.Column(db.Integer, db.ForeignKey('users.id'))
max_students = db.Column(db.Integer, nullable=False)
current_students = db.Column(db.Integer, default=0)
schedule = db.Column(db.String(255))
shift = db.Column(db.String(30), nullable=True) # Mañana, Tarde, Vespertino, Noche
virtual_link = db.Column(db.String(500), nullable=True) # Link Zoom, Meet, Teams, Moodle
active = db.Column(db.Boolean, default=True) # Changed from is_active to active
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Virtual property for compatibility
@property
def is_active(self):
return self.active
@is_active.setter
def is_active(self, value):
self.active = value
# Relationships
pass # Reservation already defines the relationship back to commission
teacher = db.relationship('User', backref='teaching_commissions', foreign_keys=[teacher_id])
# Virtual field for teacher_name
@property
def teacher_name(self):
return f"{self.teacher.first_name} {self.teacher.last_name}" if self.teacher else "TBD"
@property
def teachers(self):
"""Retorna la lista de docentes asignados.
Si teachers_association tiene registros, los retorna ordenados (Titular primero).
Si está vacía pero teacher_id está configurado, devuelve el docente titular.
"""
if hasattr(self, 'teachers_association') and self.teachers_association and len(self.teachers_association) > 0:
res = [ta.to_dict() for ta in self.teachers_association]
res.sort(key=lambda x: (not x.get('is_primary', False), x.get('role', '') != 'Titular'))
return res
if self.teacher:
return [{
'id': 0,
'commission_id': self.id,
'user_id': self.teacher.id,
'name': self.teacher.name,
'email': self.teacher.email,
'role': 'Titular',
'is_primary': True
}]
return []
@property
def teacher_names(self):
t_list = self.teachers
if not t_list:
return "Sin asignar"
return ", ".join([f"{t['name']} ({t['role']})" for t in t_list])
# Unique constraint
__table_args__ = (db.UniqueConstraint('subject_id', 'code', 'semester', 'year'),)
def to_dict(self):
"""Convert commission to dictionary"""
return {
'id': self.id,
'subject_id': self.subject_id,
'code': self.code,
'semester': self.semester,
'year': self.year,
'teacher_id': self.teacher_id,
'teacher_name': self.teacher_name,
'teachers': self.teachers,
'teacher_names': self.teacher_names,
'max_students': self.max_students,
'current_students': self.current_students,
'schedule': self.schedule,
'shift': self.shift,
'virtual_link': self.virtual_link,
'active': self.active,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'full_code': self.get_full_code(),
'subject': self.subject.to_dict() if self.subject else None
}
def get_full_code(self):
"""Get full commission code"""
subj = self.subject.code if self.subject else 'N/A'
code = self.code or ''
sem = str(self.semester or '').strip()
yr = str(self.year or '').strip()
if yr and yr in sem:
return f"{subj}-{code}-{sem}"
elif sem and yr:
return f"{subj}-{code}-{sem}{yr}"
elif sem:
return f"{subj}-{code}-{sem}"
elif yr:
return f"{subj}-{code}-{yr}"
return f"{subj}-{code}"
def __repr__(self):
return f'<Commission {self.get_full_code()}>'
class CommissionTeacher(db.Model):
"""Modelo de asignación de múltiples docentes a una comisión (Co-Docencia / Cátedra)."""
__tablename__ = 'commission_teachers'
id = db.Column(db.Integer, primary_key=True)
commission_id = db.Column(db.Integer, db.ForeignKey('commissions.id', ondelete='CASCADE'), nullable=False, index=True)
user_id = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='CASCADE'), nullable=False, index=True)
role = db.Column(db.String(50), default='Titular', nullable=False) # Titular, Adjunto, JTP, Ayudante
is_primary = db.Column(db.Boolean, default=False, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# Relationships
commission = db.relationship('Commission', backref=db.backref('teachers_association', cascade='all, delete-orphan', lazy='joined'))
teacher = db.relationship('User', backref=db.backref('commission_assignments', cascade='all, delete-orphan', lazy='joined'))
__table_args__ = (db.UniqueConstraint('commission_id', 'user_id'),)
def to_dict(self):
return {
'id': self.id,
'commission_id': self.commission_id,
'user_id': self.user_id,
'name': self.teacher.name if self.teacher else 'Docente',
'email': self.teacher.email if self.teacher else '',
'role': self.role or 'Titular',
'is_primary': self.is_primary,
'created_at': self.created_at.isoformat() if self.created_at else None
}
def __repr__(self):
return f'<CommissionTeacher Commission:{self.commission_id} User:{self.user_id} Role:{self.role}>'␍
backend/app/models/user.py
+30 -6
View File
@@ -17,15 +17,29 @@ class User(db.Model):
theme_preference = db.Column(db.String(10), default='auto', nullable=False)
role_id = db.Column(db.Integer, db.ForeignKey('roles.id', ondelete='SET NULL'), nullable=True)
# Virtual fields for future compatibility
@property
def first_name(self):
# Extended personal, contact and identity document fields
first_name = db.Column(db.String(100), nullable=True)
last_name = db.Column(db.String(100), nullable=True)
phone = db.Column(db.String(50), nullable=True)
personal_email = db.Column(db.String(255), nullable=True)
address = db.Column(db.String(255), nullable=True)
document_type = db.Column(db.String(20), default='DNI', nullable=True)
document_number = db.Column(db.String(30), nullable=True, index=True)
def get_first_name(self):
if self.first_name:
return self.first_name
return self.name.split()[0] if self.name else ''
@property
def last_name(self):
parts = self.name.split()
def get_last_name(self):
if self.last_name:
return self.last_name
parts = self.name.split() if self.name else []
return ' '.join(parts[1:]) if len(parts) > 1 else ''
def get_formatted_document(self):
from app.constants.document_types import format_document
return format_document(self.document_type, self.document_number)
# Relationships
reservations = db.relationship('Reservation', backref='user', lazy=True, cascade='all, delete-orphan')
@@ -132,8 +146,18 @@ class User(db.Model):
return {
'id': self.id,
'email': self.email,
'institutional_email': self.email,
'personal_email': self.personal_email or '',
'name': self.name,
'first_name': self.get_first_name(),
'last_name': self.get_last_name(),
'phone': self.phone or '',
'address': self.address or '',
'document_type': self.document_type or 'DNI',
'document_number': self.document_number or '',
'document_formatted': self.get_formatted_document(),
'role': self.role,
'role_id': self.role_id,
'is_active': self.is_active,
'created_at': self.created_at.isoformat() if self.created_at else None,
'last_login': self.last_login.isoformat() if self.last_login else None,
backend/app/repositories/reservation_repository.py
+19
View File
@@ -59,3 +59,22 @@ class ReservationRepository(BaseRepository[Reservation]):
return Reservation.query.filter_by(
commission_id=commission_id
).order_by(Reservation.start_time.asc()).all()
def find_teacher_conflicts(self, teacher_id: int, start_time: datetime, end_time: datetime,
exclude_reservation_id: Optional[int] = None) -> List[Reservation]:
"""
Encuentra reservas activas donde el docente ya está asignado en la misma franja horaria.
Condición de solapamiento: (start_time < existing.end_time) AND (end_time > existing.start_time).
"""
query = Reservation.query.options(
joinedload(Reservation.classroom),
joinedload(Reservation.commission)
).filter(
Reservation.user_id == teacher_id,
Reservation.status != ReservationStatus.CANCELLED.value,
Reservation.start_time < end_time,
Reservation.end_time > start_time
)
if exclude_reservation_id:
query = query.filter(Reservation.id != exclude_reservation_id)
return query.all()
backend/app/routes/api/admin.py
+307 -26
View File
@@ -4,11 +4,13 @@ import re
from app.utils.jwt_decorators import jwt_required
from app.models.user import User
from app.models.role import Role, Permission, SYSTEM_MODULES
from app.models.subject import Subject, Commission
from app.models.subject import Subject, Commission, CommissionTeacher
from app.models.career import Career
from app.models.academic_term import AcademicTerm
from app.models.milestone import MilestoneType, AcademicMilestone
from app.models.audit_log import AuditLog
from app.constants.document_types import DOCUMENT_TYPES, validate_document, format_document
from app.services.enrollment_service import EnrollmentService
from app import db
api_admin_bp = Blueprint('api_admin', __name__)
@@ -22,7 +24,14 @@ def get_users():
query = User.query
if search:
query = query.filter((User.name.ilike(f'%{search}%')) | (User.email.ilike(f'%{search}%')))
query = query.filter(
(User.name.ilike(f'%{search}%')) |
(User.email.ilike(f'%{search}%')) |
(User.personal_email.ilike(f'%{search}%')) |
(User.document_number.ilike(f'%{search}%')) |
(User.first_name.ilike(f'%{search}%')) |
(User.last_name.ilike(f'%{search}%'))
)
if role_id:
query = query.filter(User.role_id == role_id)
if status == 'active':
@@ -35,19 +44,12 @@ def get_users():
users_data = []
for u in users:
users_data.append({
'id': u.id,
'name': u.name,
'first_name': u.name.split(' ')[0] if u.name else 'U',
'email': u.email,
'is_active': u.is_active,
'role': u.role,
'role_obj': {
'id': u.role_obj.id,
'name': u.role_obj.name
} if u.role_obj else {'id': 1, 'name': u.role or 'Admin'},
'last_login': getattr(u, 'last_login', None)
})
d = u.to_dict()
d['role_obj'] = {
'id': u.role_obj.id,
'name': u.role_obj.name
} if u.role_obj else {'id': 1, 'name': u.role or 'Admin'}
users_data.append(d)
roles_data = [{'id': r.id, 'name': r.name} for r in roles]
@@ -55,7 +57,8 @@ def get_users():
'status': 'success',
'total': len(users_data),
'users': users_data,
'roles': roles_data
'roles': roles_data,
'document_types': DOCUMENT_TYPES
}), 200
@api_admin_bp.route('/roles', methods=['GET'])
@@ -213,6 +216,8 @@ def get_commissions():
'year': getattr(c, 'year', 2026) or 2026,
'teacher_id': c.teacher_id,
'teacher_name': c.teacher_name if hasattr(c, 'teacher_name') else (c.teacher.name if c.teacher else None),
'teachers': c.teachers if hasattr(c, 'teachers') else [],
'teacher_names': c.teacher_names if hasattr(c, 'teacher_names') else (c.teacher_name if hasattr(c, 'teacher_name') else 'Sin asignar'),
'schedule': c.schedule or 'A coordinar',
'shift': c.shift or 'Mañana',
'max_students': c.max_students or 35,
@@ -248,19 +253,47 @@ def get_commissions():
def create_user():
data = request.get_json(silent=True) or request.form.to_dict() or {}
email = data.get('email', '').strip().lower()
first_name = data.get('first_name', '').strip()
last_name = data.get('last_name', '').strip()
name = data.get('name', '').strip()
if not name and (first_name or last_name):
name = f"{first_name} {last_name}".strip()
elif name and not first_name:
parts = name.split()
first_name = parts[0] if parts else ''
last_name = ' '.join(parts[1:]) if len(parts) > 1 else ''
password = data.get('password', '').strip()
role_id = data.get('role_id')
is_active = data.get('is_active', True)
if isinstance(is_active, str):
is_active = is_active.lower() in ['true', '1', 'on']
phone = data.get('phone', '').strip()
personal_email = data.get('personal_email', '').strip().lower()
address = data.get('address', '').strip()
document_type = data.get('document_type', 'DNI').strip().upper()
document_number = data.get('document_number', '').strip()
if not email or not name:
return jsonify({'error': 'ValidationError', 'message': 'El nombre y el email son obligatorios.'}), 400
if User.query.filter(User.email.ilike(email)).first():
return jsonify({'error': 'Conflict', 'message': f'Ya existe un usuario con el email {email}.'}), 409
if document_number:
is_valid, clean_doc, err_msg = validate_document(document_type, document_number)
if not is_valid:
return jsonify({'error': 'ValidationError', 'message': err_msg}), 400
document_number = clean_doc
existing_doc = User.query.filter(
User.document_type == document_type,
User.document_number == document_number
).first()
if existing_doc:
return jsonify({'error': 'Conflict', 'message': f'Ya existe un usuario con {document_type} {document_number}.'}), 409
role_obj = None
if role_id:
role_obj = Role.query.get(int(role_id))
@@ -268,7 +301,14 @@ def create_user():
user = User(
email=email,
personal_email=personal_email,
name=name,
first_name=first_name,
last_name=last_name,
phone=phone,
address=address,
document_type=document_type,
document_number=document_number,
role=role_name,
role_id=int(role_id) if role_id else None,
is_active=bool(is_active)
@@ -280,7 +320,7 @@ def create_user():
return jsonify({
'status': 'success',
'message': 'Usuario creado exitosamente.',
'user': {'id': user.id, 'name': user.name, 'email': user.email, 'role': user.role, 'is_active': user.is_active}
'user': user.to_dict()
}), 201
@api_admin_bp.route('/users/<int:id>', methods=['PUT'])
@@ -289,8 +329,48 @@ def update_user(id):
user = User.query.get_or_404(id)
data = request.get_json(silent=True) or request.form.to_dict() or {}
if 'name' in data and data['name']:
user.name = data['name'].strip()
first_name = data.get('first_name', user.first_name or '').strip()
last_name = data.get('last_name', user.last_name or '').strip()
name = data.get('name', '').strip()
if 'first_name' in data or 'last_name' in data:
user.first_name = first_name
user.last_name = last_name
user.name = f"{first_name} {last_name}".strip()
elif name:
user.name = name
parts = name.split()
user.first_name = parts[0] if parts else ''
user.last_name = ' '.join(parts[1:]) if len(parts) > 1 else ''
if 'phone' in data:
user.phone = (data['phone'] or '').strip()
if 'personal_email' in data:
user.personal_email = (data['personal_email'] or '').strip().lower()
if 'address' in data:
user.address = (data['address'] or '').strip()
if 'document_type' in data or 'document_number' in data:
doc_type = data.get('document_type', user.document_type or 'DNI').strip().upper()
doc_num = data.get('document_number', user.document_number or '').strip()
if doc_num:
is_valid, clean_doc, err_msg = validate_document(doc_type, doc_num)
if not is_valid:
return jsonify({'error': 'ValidationError', 'message': err_msg}), 400
doc_num = clean_doc
existing_doc = User.query.filter(
User.document_type == doc_type,
User.document_number == doc_num,
User.id != id
).first()
if existing_doc:
return jsonify({'error': 'Conflict', 'message': f'El documento {doc_type} {doc_num} ya está asignado a otro usuario.'}), 409
user.document_type = doc_type
user.document_number = doc_num
if 'email' in data and data['email']:
email = data['email'].strip().lower()
existing = User.query.filter(User.email.ilike(email), User.id != id).first()
@@ -312,7 +392,15 @@ def update_user(id):
return jsonify({
'status': 'success',
'message': 'Usuario actualizado correctamente.',
'user': {'id': user.id, 'name': user.name, 'email': user.email, 'role': user.role, 'is_active': user.is_active}
'user': user.to_dict()
}), 200
@api_admin_bp.route('/users/document-types', methods=['GET'])
@jwt_required
def get_document_types():
return jsonify({
'status': 'success',
'document_types': DOCUMENT_TYPES
}), 200
@api_admin_bp.route('/users/<int:id>/toggle', methods=['POST'])
@@ -598,15 +686,18 @@ def delete_subject(id):
subject.is_active = False
db.session.commit()
return jsonify({
'status': 'success',
'message': 'La asignatura tiene comisiones activas. Se ha desactivado en su lugar.'
'status': 'deactivated',
'action': 'deactivated',
'message': f'La asignatura "{subject.name}" tiene {len(subject.commissions)} comisión(es) vinculada(s). Se ha desactivado en su lugar para proteger el historial académico.'
}), 200
name = subject.name
db.session.delete(subject)
db.session.commit()
return jsonify({
'status': 'success',
'message': 'Asignatura eliminada permanentemente.'
'status': 'deleted',
'action': 'deleted',
'message': f'Asignatura "{name}" eliminada permanentemente del sistema.'
}), 200
# ---------------------------------------------------------
@@ -659,6 +750,10 @@ def create_commission():
if virtual_link:
existing.virtual_link = virtual_link
existing.active = bool(active)
if teacher_id:
ct = CommissionTeacher.query.filter_by(commission_id=existing.id, user_id=teacher_id).first()
if not ct:
db.session.add(CommissionTeacher(commission_id=existing.id, user_id=teacher_id, role='Titular', is_primary=True))
db.session.commit()
return jsonify({
'status': 'success',
@@ -679,6 +774,19 @@ def create_commission():
active=bool(active)
)
db.session.add(commission)
db.session.flush()
if teacher_id:
db.session.add(CommissionTeacher(commission_id=commission.id, user_id=teacher_id, role='Titular', is_primary=True))
# Soporte para docentes adicionales al crear comisión
extra_teachers = data.get('teacher_ids') or data.get('teachers') or []
for et in extra_teachers:
u_id = et.get('user_id') if isinstance(et, dict) else et
u_role = et.get('role', 'Adjunto') if isinstance(et, dict) else 'Adjunto'
if u_id and int(u_id) != teacher_id:
db.session.add(CommissionTeacher(commission_id=commission.id, user_id=int(u_id), role=u_role, is_primary=False))
db.session.commit()
return jsonify({
@@ -688,6 +796,16 @@ def create_commission():
}), 201
@api_admin_bp.route('/commissions/<int:id>', methods=['GET'])
@jwt_required
def get_commission_detail(id):
comm = Commission.query.get_or_404(id)
return jsonify({
'status': 'success',
'commission': comm.to_dict()
}), 200
@api_admin_bp.route('/commissions/<int:id>', methods=['PUT'])
@jwt_required
def update_commission(id):
@@ -703,6 +821,10 @@ def update_commission(id):
if 'teacher_id' in data:
t_id = data['teacher_id']
comm.teacher_id = int(t_id) if t_id else None
if comm.teacher_id:
ct = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=comm.teacher_id).first()
if not ct:
db.session.add(CommissionTeacher(commission_id=comm.id, user_id=comm.teacher_id, role='Titular', is_primary=True))
if 'max_students' in data and data['max_students']:
comm.max_students = int(data['max_students'])
if 'schedule' in data:
@@ -740,17 +862,176 @@ def toggle_commission(id):
def assign_commission_teacher(id):
comm = Commission.query.get_or_404(id)
data = request.get_json(silent=True) or request.form.to_dict() or {}
t_id = data.get('teacher_id')
comm.teacher_id = int(t_id) if t_id else None
t_id = data.get('teacher_id') or data.get('user_id')
role = (data.get('role') or 'Titular').strip()
if t_id:
t_id = int(t_id)
comm.teacher_id = t_id
ct = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=t_id).first()
if not ct:
ct = CommissionTeacher(commission_id=comm.id, user_id=t_id, role=role, is_primary=True)
db.session.add(ct)
else:
ct.role = role
ct.is_primary = True
else:
comm.teacher_id = None
db.session.commit()
return jsonify({
'status': 'success',
'id': comm.id,
'teacher_id': comm.teacher_id,
'teacher_name': comm.teacher_name,
'teachers': comm.teachers,
'message': 'Docente asignado correctamente.'
}), 200
@api_admin_bp.route('/commissions/<int:id>/teachers', methods=['POST'])
@jwt_required
def add_commission_teacher(id):
comm = Commission.query.get_or_404(id)
data = request.get_json(silent=True) or request.form.to_dict() or {}
user_id = data.get('user_id') or data.get('teacher_id')
if not user_id:
return jsonify({'error': 'ValidationError', 'message': 'El docente es requerido.'}), 400
user = User.query.get(int(user_id))
if not user:
return jsonify({'error': 'NotFound', 'message': f'El usuario docente con ID {user_id} no existe.'}), 404
role = (data.get('role') or 'Adjunto').strip()
is_primary = data.get('is_primary', False)
if isinstance(is_primary, str):
is_primary = is_primary.lower() in ['true', '1', 'on']
existing = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=user.id).first()
if existing:
existing.role = role
existing.is_primary = bool(is_primary)
else:
if is_primary or comm.teacher_id is None:
comm.teacher_id = user.id
is_primary = True
ct = CommissionTeacher(
commission_id=comm.id,
user_id=user.id,
role=role,
is_primary=bool(is_primary)
)
db.session.add(ct)
db.session.commit()
return jsonify({
'status': 'success',
'message': f'{user.name} asignado/a como {role} en la comisión.',
'teachers': comm.teachers,
'commission': comm.to_dict()
}), 200
@api_admin_bp.route('/commissions/<int:id>/teachers/<int:user_id>', methods=['DELETE'])
@jwt_required
def remove_commission_teacher(id, user_id):
comm = Commission.query.get_or_404(id)
ct = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=user_id).first()
if ct:
db.session.delete(ct)
# Si era el docente titular, reasignar a otro docente disponible o dejar en None
if comm.teacher_id == user_id:
other = CommissionTeacher.query.filter(CommissionTeacher.commission_id == comm.id, CommissionTeacher.user_id != user_id).first()
comm.teacher_id = other.user_id if other else None
if other:
other.is_primary = True
db.session.commit()
return jsonify({
'status': 'success',
'message': 'Docente desvinculado de la comisión correctamente.',
'teachers': comm.teachers,
'commission': comm.to_dict()
}), 200
# ---------------------------------------------------------
# COMMISSION STUDENT ENROLLMENTS (FASE 2 MVP)
# ---------------------------------------------------------
@api_admin_bp.route('/commissions/<int:id>/enrollments', methods=['GET'])
@jwt_required
def get_commission_enrollments(id):
comm = Commission.query.get_or_404(id)
enrollments = EnrollmentService.list_enrollments(comm.id)
return jsonify({
'commission_id': comm.id,
'commission_code': comm.code,
'count': len(enrollments),
'max_students': comm.max_students,
'current_students': comm.current_students,
'enrollments': enrollments
}), 200
@api_admin_bp.route('/commissions/<int:id>/enrollments', methods=['POST'])
@jwt_required
def enroll_commission_student(id):
data = request.get_json(silent=True) or {}
student_id = data.get('student_id')
notes = data.get('notes')
allow_same_day_exception = bool(data.get('allow_same_day_exception', False))
exception_reason = data.get('exception_reason')
if not student_id:
return jsonify({'error': 'BadRequest', 'message': 'student_id es obligatorio.'}), 400
try:
enrollment = EnrollmentService.enroll_student(
commission_id=id,
student_id=int(student_id),
notes=notes,
allow_same_day_exception=allow_same_day_exception,
exception_reason=exception_reason
)
return jsonify({
'status': 'success',
'message': 'Estudiante matriculado exitosamente en la comisión.',
'enrollment': enrollment.to_dict()
}), 201
except ValueError as e:
return jsonify({'error': 'ConflictOrValidation', 'message': str(e)}), 409
@api_admin_bp.route('/commissions/<int:id>/enrollments/<int:student_id>', methods=['DELETE'])
@jwt_required
def unenroll_commission_student(id, student_id):
success = EnrollmentService.unenroll_student(commission_id=id, student_id=student_id)
if not success:
return jsonify({'error': 'NotFound', 'message': 'Matrícula no encontrada para este estudiante.'}), 404
return jsonify({
'status': 'success',
'message': 'Estudiante desvinculado de la comisión exitosamente.'
}), 200
@api_admin_bp.route('/commissions/<int:id>/enrollments/<int:student_id>', methods=['PUT'])
@jwt_required
def update_commission_enrollment(id, student_id):
data = request.get_json(silent=True) or {}
status = data.get('status', 'activo')
notes = data.get('notes')
enrollment = EnrollmentService.update_enrollment_status(
commission_id=id,
student_id=student_id,
status=status,
notes=notes
)
if not enrollment:
return jsonify({'error': 'NotFound', 'message': 'Matrícula no encontrada.'}), 404
return jsonify({
'status': 'success',
'message': 'Estado de matrícula actualizado.',
'enrollment': enrollment.to_dict()
}), 200
# ---------------------------------------------------------
# ACADEMIC TERMS CRUD
# ---------------------------------------------------------
backend/app/routes/api/auth.py
+6 -2
View File
@@ -15,7 +15,9 @@ def login():
"""
Endpoint de autenticación para obtener par de tokens (Access + Refresh).
"""
data = request.get_json(silent=True) or {}
data = request.get_json(silent=True)
if not isinstance(data, dict):
return jsonify({'error': 'BadRequest', 'message': 'El cuerpo de la solicitud debe ser un objeto JSON.'}), 400
try:
dto = LoginDTO(**data)
except ValidationError as e:
@@ -56,7 +58,9 @@ def refresh():
"""
Renovación silenciosa del Access Token utilizando el Refresh Token.
"""
data = request.get_json(silent=True) or {}
data = request.get_json(silent=True)
if not isinstance(data, dict):
data = {}
refresh_token = data.get('refresh_token') or request.cookies.get('refresh_token')
if not refresh_token:
backend/app/routes/api/openapi.py
+242
View File
@@ -0,0 +1,242 @@
from flask import Blueprint, jsonify
api_openapi_bp = Blueprint('api_openapi', __name__, url_prefix='/api/v1')
OPENAPI_SPEC = {
"openapi": "3.0.3",
"info": {
"title": "Edu-Space REST API",
"description": "Enterprise Academic & Physical Space Management REST API with JWT Authentication and Role-Based Access Control.",
"version": "1.0.0"
},
"servers": [
{
"url": "http://127.0.0.1:5000",
"description": "Local Flask Backend Server"
}
],
"components": {
"securitySchemes": {
"bearerAuth": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT"
}
},
"schemas": {
"LoginRequest": {
"type": "object",
"required": ["email", "password"],
"properties": {
"email": {
"type": "string",
"format": "email",
"example": "admin@eduspace.com"
},
"password": {
"type": "string",
"format": "password",
"example": "Admin123!"
}
}
},
"LoginResponse": {
"type": "object",
"properties": {
"access_token": {"type": "string"},
"refresh_token": {"type": "string"},
"token_type": {"type": "string", "example": "Bearer"},
"expires_in": {"type": "integer"},
"user": {"type": "object"}
}
},
"ErrorResponse": {
"type": "object",
"properties": {
"error": {"type": "string"},
"message": {"type": "string"}
}
},
"UserResponse": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"email": {"type": "string"},
"first_name": {"type": "string"},
"last_name": {"type": "string"},
"role": {"type": "string"},
"is_active": {"type": "boolean"}
}
},
"SubjectResponse": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"code": {"type": "string"},
"career": {"type": "string"}
}
},
"ClassroomResponse": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"capacity": {"type": "integer"},
"building": {"type": "string"},
"floor": {"type": "string"}
}
}
}
},
"paths": {
"/api/v1/auth/login": {
"post": {
"summary": "Authenticate user and issue JWT tokens",
"tags": ["Authentication"],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/LoginRequest"
}
}
}
},
"responses": {
"200": {
"description": "Authentication successful",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/LoginResponse"
}
}
}
},
"400": {
"description": "Validation error",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"}
}
}
},
"401": {
"description": "Unauthorized / Bad credentials",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"}
}
}
}
}
}
},
"/api/v1/auth/me": {
"get": {
"summary": "Get authenticated user profile",
"tags": ["Authentication"],
"security": [{"bearerAuth": []}],
"responses": {
"200": {
"description": "Current user profile",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/UserResponse"}
}
}
},
"401": {
"description": "Missing or invalid token",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"}
}
}
}
}
}
},
"/api/v1/classrooms": {
"get": {
"summary": "List all physical classrooms and facilities",
"tags": ["Classrooms"],
"security": [{"bearerAuth": []}],
"responses": {
"200": {
"description": "List of classrooms",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {"$ref": "#/components/schemas/ClassroomResponse"}
}
}
}
}
}
}
},
"/api/v1/admin/users": {
"get": {
"summary": "List all users (Admin only)",
"tags": ["Admin - Users"],
"security": [{"bearerAuth": []}],
"responses": {
"200": {
"description": "List of users",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {"$ref": "#/components/schemas/UserResponse"}
}
}
}
},
"403": {
"description": "Forbidden - Requires admin role"
}
}
}
},
"/api/v1/admin/subjects": {
"get": {
"summary": "List all academic subjects (Admin only)",
"tags": ["Admin - Academic"],
"security": [{"bearerAuth": []}],
"responses": {
"200": {
"description": "List of subjects",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {"$ref": "#/components/schemas/SubjectResponse"}
}
}
}
}
}
}
},
"/api/v1/openapi.json": {
"get": {
"summary": "Get OpenAPI 3.0.3 specification JSON",
"tags": ["Documentation"],
"responses": {
"200": {
"description": "OpenAPI Specification"
}
}
}
}
}
}
@api_openapi_bp.route('/openapi.json', methods=['GET'])
def get_openapi_spec():
"""Serves the OpenAPI 3.0 specification for Schemathesis / DAST and Swagger UI."""
return jsonify(OPENAPI_SPEC)
backend/app/routes/api/reservations.py
+45
View File
@@ -245,3 +245,48 @@ def confirm_reservation(reservation_id: int):
return jsonify({'message': f'Reserva {reservation_id} confirmada correctamente.'}), 200
@api_reservations_bp.route('/check-conflicts', methods=['POST'])
@jwt_required
def check_conflicts_endpoint():
"""
Evalúa la matriz completa de conflictos (físico, docente y aforo) en tiempo real.
"""
data = request.get_json(silent=True)
if not isinstance(data, dict):
return jsonify({'error': 'BadRequest', 'message': 'El cuerpo de la solicitud debe ser un objeto JSON.'}), 400
classroom_id = data.get('classroom_id')
start_str = data.get('start_time')
end_str = data.get('end_time')
if not classroom_id or not start_str or not end_str:
return jsonify({'error': 'BadRequest', 'message': 'classroom_id, start_time y end_time son obligatorios.'}), 400
try:
start_time = datetime.fromisoformat(str(start_str).replace('Z', '+00:00'))
end_time = datetime.fromisoformat(str(end_str).replace('Z', '+00:00'))
except ValueError:
return jsonify({'error': 'BadRequest', 'message': 'Formato inválido de fecha/hora (ISO 8601 esperado).'}), 400
teacher_id = data.get('teacher_id')
commission_id = data.get('commission_id')
expected_attendees = data.get('expected_attendees')
allow_co_teaching = bool(data.get('allow_co_teaching_exception', False))
exclude_id = data.get('exclude_id')
try:
matrix = reservation_service.check_full_conflicts_matrix(
classroom_id=int(classroom_id),
start_time=start_time,
end_time=end_time,
teacher_id=int(teacher_id) if teacher_id else None,
commission_id=int(commission_id) if commission_id else None,
expected_attendees=int(expected_attendees) if expected_attendees is not None else None,
allow_co_teaching_exception=allow_co_teaching,
exclude_id=int(exclude_id) if exclude_id else None
)
return jsonify(matrix), 200
except ValueError as e:
return jsonify({'error': 'ValidationError', 'message': str(e)}), 400
backend/app/schemas/reservation_dto.py
+2 -1
View File
@@ -22,7 +22,7 @@ class ReservationBaseDTO(BaseModel):
return self
class ReservationCreateDTO(ReservationBaseDTO):
pass
allow_co_teaching_exception: Optional[bool] = Field(default=False, description="Excepción justificada si hay co-docencia")
class ReservationUpdateDTO(BaseModel):
classroom_id: Optional[int] = None
@@ -34,6 +34,7 @@ class ReservationUpdateDTO(BaseModel):
virtual_link: Optional[str] = None
notes: Optional[str] = None
status: Optional[str] = None
allow_co_teaching_exception: Optional[bool] = False
class ReservationResponseDTO(ReservationBaseDTO):
id: int
backend/app/services/enrollment_service.py
+220
View File
@@ -0,0 +1,220 @@
import re
from typing import Optional, List, Set, Dict, Any
from datetime import datetime
from app import db
from app.models.user import User
from app.models.subject import Commission, Subject
from app.models.enrollment import StudentEnrollment
from app.models.reservation import Reservation
WEEKDAYS_MAP = {
0: 'lunes',
1: 'martes',
2: 'miercoles',
3: 'jueves',
4: 'viernes',
5: 'sabado',
6: 'domingo'
}
SPANISH_DAYS = ['lunes', 'martes', 'miercoles', 'miércoles', 'jueves', 'viernes', 'sabado', 'sábado', 'domingo']
class EnrollmentService:
"""
Servicio de matriculación y aplicación de reglas de negocio académicas (Fase 2 MVP):
- Control estricto de aforo y cupo de comisión.
- Prevención de doble inscripción.
- Regla de Restricción Diaria del Alumno (máximo 1 asignatura regular por día calendario).
- Soporte de excepciones automáticas por curso corto y excepciones autorizadas de Bedelía.
"""
@staticmethod
def normalize_day(day_str: str) -> str:
d = day_str.lower().strip()
if 'miér' in d or 'mier' in d:
return 'miercoles'
if 'sáb' in d or 'sab' in d:
return 'sabado'
return d
@classmethod
def get_commission_days(cls, commission: Commission) -> Set[str]:
"""Extrae el conjunto de días de cursada de la comisión desde schedule y reservas activas."""
days: Set[str] = set()
# 1. Analizar texto del campo schedule
if commission.schedule:
sched_lower = commission.schedule.lower()
for d in SPANISH_DAYS:
if re.search(r'\b' + re.escape(d) + r'\b', sched_lower):
days.add(cls.normalize_day(d))
# 2. Analizar reservas reales de la comisión
reservations = Reservation.query.filter(
Reservation.commission_id == commission.id,
Reservation.status != 'CANCELLED'
).all()
for r in reservations:
if r.start_time:
weekday_idx = r.start_time.weekday()
if weekday_idx in WEEKDAYS_MAP:
days.add(WEEKDAYS_MAP[weekday_idx])
return days
@classmethod
def list_enrollments(cls, commission_id: int) -> List[Dict[str, Any]]:
"""Lista todos los alumnos matriculados en una comisión con información extendida."""
enrollments = StudentEnrollment.query.filter_by(commission_id=commission_id).all()
result = []
for enr in enrollments:
data = enr.to_dict()
if enr.student:
data['student'] = {
'id': enr.student.id,
'name': enr.student.name,
'email': enr.student.email,
'personal_email': getattr(enr.student, 'personal_email', None),
'phone': getattr(enr.student, 'phone', None),
'document_type': getattr(enr.student, 'document_type', None),
'document_number': getattr(enr.student, 'document_number', None)
}
result.append(data)
return result
@classmethod
def enroll_student(cls, commission_id: int, student_id: int,
notes: Optional[str] = None,
allow_same_day_exception: bool = False,
exception_reason: Optional[str] = None) -> StudentEnrollment:
"""
Matricula a un estudiante en una comisión aplicando la regla de restricción diaria.
"""
commission = Commission.query.get(commission_id)
if not commission or not commission.active:
raise ValueError(f"La comisión #{commission_id} no existe o no se encuentra activa.")
student = User.query.get(student_id)
if not student or not student.is_active:
raise ValueError(f"El alumno #{student_id} no existe o su cuenta se encuentra inactiva.")
# Verificar si ya está matriculado
existing = StudentEnrollment.query.filter_by(commission_id=commission_id, student_id=student_id).first()
if existing:
if existing.status == 'activo':
raise ValueError(f"El alumno {student.name} ya está matriculado activamente en esta comisión.")
# Si estaba retirado, se reactiva
existing.status = 'activo'
existing.notes = notes or existing.notes
existing.allow_same_day_exception = allow_same_day_exception
existing.exception_reason = exception_reason if allow_same_day_exception else None
commission.current_students = (commission.current_students or 0) + 1
db.session.commit()
return existing
# Verificar cupo máximo
if commission.max_students and (commission.current_students or 0) >= commission.max_students:
raise ValueError(f"La comisión ha alcanzado su cupo máximo de {commission.max_students} estudiantes.")
# ─── REGLA DE RESTRICCIÓN DIARIA DE CURSADA (ESTADIO 2.3) ───
target_subject = commission.subject
is_target_short_course = bool(target_subject and target_subject.is_short_course)
target_days = cls.get_commission_days(commission)
# Buscar otras inscripciones activas del estudiante
active_enrollments = StudentEnrollment.query.filter_by(
student_id=student_id,
status='activo'
).all()
for enr in active_enrollments:
other_comm = enr.commission
if not other_comm or not other_comm.active:
continue
# Verificar si coinciden en el mismo ciclo/semestre/año
same_term = (
other_comm.year == commission.year and
other_comm.semester == commission.semester
)
if not same_term:
continue
other_subject = other_comm.subject
is_other_short_course = bool(other_subject and other_subject.is_short_course)
# Ambas son materias regulares: no pueden cursarse el mismo día sin excepción
if not is_target_short_course and not is_other_short_course:
other_days = cls.get_commission_days(other_comm)
overlapping_days = target_days.intersection(other_days)
if overlapping_days:
if allow_same_day_exception:
# Excepción otorgada expresamente por Bedelía
if not exception_reason:
exception_reason = "Autorización expresa de Bedelía para cursada simultánea en el mismo día."
else:
days_display = ", ".join([d.capitalize() for d in overlapping_days])
other_name = other_subject.name if other_subject else other_comm.code
raise ValueError(
f"Restricción de cursada diaria: El alumno ya cursa la materia regular '{other_name}' "
f"el día {days_display}. La normativa de UniCABA prohíbe cursar dos materias regulares "
f"el mismo día calendario salvo curso corto o autorización explícita de Bedelía."
)
enrollment = StudentEnrollment(
student_id=student_id,
commission_id=commission_id,
status='activo',
notes=notes,
allow_same_day_exception=allow_same_day_exception,
exception_reason=exception_reason if allow_same_day_exception else None
)
db.session.add(enrollment)
commission.current_students = (commission.current_students or 0) + 1
db.session.commit()
return enrollment
@classmethod
def unenroll_student(cls, commission_id: int, student_id: int) -> bool:
"""Da de baja o retira a un alumno de una comisión."""
enrollment = StudentEnrollment.query.filter_by(
commission_id=commission_id,
student_id=student_id
).first()
if not enrollment:
return False
commission = Commission.query.get(commission_id)
if commission and (commission.current_students or 0) > 0:
commission.current_students -= 1
db.session.delete(enrollment)
db.session.commit()
return True
@classmethod
def update_enrollment_status(cls, commission_id: int, student_id: int, status: str,
notes: Optional[str] = None) -> Optional[StudentEnrollment]:
"""Actualiza el estado de la matrícula (activo, condicional, retirado)."""
enrollment = StudentEnrollment.query.filter_by(
commission_id=commission_id,
student_id=student_id
).first()
if not enrollment:
return None
old_status = enrollment.status
enrollment.status = status
if notes is not None:
enrollment.notes = notes
commission = Commission.query.get(commission_id)
if commission:
if old_status == 'activo' and status in ['retirado']:
commission.current_students = max(0, (commission.current_students or 1) - 1)
elif old_status in ['retirado'] and status == 'activo':
commission.current_students = (commission.current_students or 0) + 1
db.session.commit()
return enrollment
backend/app/services/reservation_service.py
+123 -25
View File
@@ -1,7 +1,8 @@
from typing import Optional, List
from typing import Optional, List, Dict, Any
from datetime import datetime
from app.models.reservation import Reservation, ReservationStatus
from app.models.classroom import Classroom
from app.models.subject import Commission, CommissionTeacher
from app.repositories.reservation_repository import ReservationRepository
from app.repositories.classroom_repository import ClassroomRepository
from app.schemas.reservation_dto import ReservationCreateDTO, ReservationUpdateDTO
@@ -38,23 +39,116 @@ class ReservationService:
exclude_reservation_id=exclude_id
)
def check_teacher_conflicts(self, teacher_id: int, start_time: datetime, end_time: datetime,
exclude_id: Optional[int] = None) -> List[Reservation]:
"""Verifica si el docente ya tiene otra reserva en la misma franja horaria."""
if not teacher_id:
return []
return self.reservation_repo.find_teacher_conflicts(
teacher_id=teacher_id,
start_time=start_time,
end_time=end_time,
exclude_reservation_id=exclude_id
)
def check_full_conflicts_matrix(self, classroom_id: int, start_time: datetime, end_time: datetime,
teacher_id: Optional[int] = None,
commission_id: Optional[int] = None,
expected_attendees: Optional[int] = None,
allow_co_teaching_exception: bool = False,
exclude_id: Optional[int] = None) -> Dict[str, Any]:
"""
Matriz completa de validación de conflictos (Fase 2 MVP):
1. Conflicto físico de aula (bloqueo estricto para aulas físicas).
2. Conflicto docente (detección con soporte de co-docencia).
3. Conflicto de aforo (capacidad física).
"""
classroom = self.classroom_repo.get_by_id(classroom_id)
if not classroom or not classroom.is_active:
raise ValueError(f"El aula con ID {classroom_id} no existe o no se encuentra activa.")
commission = None
if commission_id:
commission = Commission.query.get(commission_id)
if not teacher_id and commission:
teacher_id = commission.teacher_id
# 1. Conflicto Físico
physical_conflicts = []
if not classroom.is_virtual:
physical_conflicts = self.check_conflicts(classroom_id, start_time, end_time, exclude_id=exclude_id)
# 2. Conflicto Docente y Co-docencia
teacher_conflicts = []
has_co_teaching_coverage = False
co_teachers_list = []
if teacher_id:
teacher_conflicts = self.check_teacher_conflicts(teacher_id, start_time, end_time, exclude_id=exclude_id)
if teacher_conflicts and commission:
# Comprobar si la comisión cuenta con equipo docente de respaldo
co_teachers = CommissionTeacher.query.filter_by(commission_id=commission.id).all()
other_teachers = [ct for ct in co_teachers if ct.user_id != teacher_id]
if other_teachers:
has_co_teaching_coverage = True
co_teachers_list = [
{'id': ct.user_id, 'name': ct.teacher.name if ct.teacher else 'Docente', 'role': ct.role}
for ct in other_teachers
]
# 3. Conflicto de Aforo
has_capacity_conflict = False
capacity_message = None
if not classroom.is_virtual and expected_attendees is not None:
if expected_attendees > classroom.capacity:
has_capacity_conflict = True
capacity_message = f"La cantidad esperada ({expected_attendees}) supera la capacidad del aula ({classroom.capacity})."
# Evaluación de Bloqueos
block_reasons = []
if physical_conflicts:
c_strs = [f"#{c.id} ({c.start_time.strftime('%H:%M')} a {c.end_time.strftime('%H:%M')})" for c in physical_conflicts]
block_reasons.append(f"Conflicto de horario en aula {classroom.code}: se solapa con las reservas {', '.join(c_strs)}.")
if has_capacity_conflict:
block_reasons.append(f"Conflicto de aforo: {capacity_message}")
if teacher_conflicts:
if has_co_teaching_coverage or allow_co_teaching_exception:
# Cobertura justificada por equipo de co-docencia
pass
else:
t_strs = [f"#{c.id} ({c.start_time.strftime('%H:%M')} a {c.end_time.strftime('%H:%M')})" for c in teacher_conflicts]
block_reasons.append(f"Conflicto de horario docente: el profesor ya tiene clases asignadas en {', '.join(t_strs)} y la comisión no cuenta con equipo de co-docencia.")
is_blocked = len(block_reasons) > 0
return {
'is_valid': not is_blocked,
'is_blocked': is_blocked,
'block_reasons': block_reasons,
'physical_conflicts': [c.to_dict() for c in physical_conflicts],
'teacher_conflicts': [c.to_dict() for c in teacher_conflicts],
'has_co_teaching_coverage': has_co_teaching_coverage,
'co_teachers': co_teachers_list,
'has_capacity_conflict': has_capacity_conflict,
'classroom': classroom.to_dict()
}
def create_reservation(self, dto: ReservationCreateDTO) -> Reservation:
"""
Crea una nueva reserva aplicando validaciones de aforo y detección de colisiones.
Crea una nueva reserva aplicando la matriz integral de conflictos.
"""
classroom = self.classroom_repo.get_by_id(dto.classroom_id)
if not classroom or not classroom.is_active:
raise ValueError(f"El aula con ID {dto.classroom_id} no existe o no se encuentra activa.")
# Detección de colisiones para aulas físicas
if not classroom.is_virtual:
conflicts = self.check_conflicts(dto.classroom_id, dto.start_time, dto.end_time)
if conflicts:
conflict_details = ", ".join([f"#{c.id} ({c.start_time.strftime('%H:%M')} a {c.end_time.strftime('%H:%M')})" for c in conflicts])
raise ValueError(f"Conflicto de horario en {classroom.code}: se solapa con las reservas {conflict_details}.")
if dto.expected_attendees > classroom.capacity:
raise ValueError(f"La cantidad de alumnos ({dto.expected_attendees}) supera la capacidad del aula ({classroom.capacity}).")
matrix = self.check_full_conflicts_matrix(
classroom_id=dto.classroom_id,
start_time=dto.start_time,
end_time=dto.end_time,
teacher_id=dto.user_id,
commission_id=dto.commission_id,
expected_attendees=dto.expected_attendees,
allow_co_teaching_exception=getattr(dto, 'allow_co_teaching_exception', False)
)
if matrix['is_blocked']:
raise ValueError(" | ".join(matrix['block_reasons']))
reservation = Reservation(
classroom_id=dto.classroom_id,
@@ -72,7 +166,7 @@ class ReservationService:
return self.reservation_repo.save(reservation)
def update_reservation(self, reservation_id: int, dto: ReservationUpdateDTO) -> Reservation:
"""Actualiza una reserva existente verificando disponibilidad horaria."""
"""Actualiza una reserva existente verificando disponibilidad horaria y matriz de conflictos."""
reservation = self.reservation_repo.get_by_id(reservation_id)
if not reservation:
raise ValueError(f"Reserva con ID {reservation_id} no encontrada.")
@@ -80,16 +174,20 @@ class ReservationService:
classroom_id = dto.classroom_id or reservation.classroom_id
start_time = dto.start_time or reservation.start_time
end_time = dto.end_time or reservation.end_time
expected_attendees = dto.expected_attendees if dto.expected_attendees is not None else reservation.expected_attendees
classroom = self.classroom_repo.get_by_id(classroom_id)
if not classroom or not classroom.is_active:
raise ValueError(f"El aula con ID {classroom_id} no existe o no se encuentra activa.")
if not classroom.is_virtual:
conflicts = self.check_conflicts(classroom_id, start_time, end_time, exclude_id=reservation.id)
if conflicts:
conflict_details = ", ".join([f"#{c.id} ({c.start_time.strftime('%H:%M')} - {c.end_time.strftime('%H:%M')})" for c in conflicts])
raise ValueError(f"Conflicto de horario en {classroom.code} con: {conflict_details}.")
matrix = self.check_full_conflicts_matrix(
classroom_id=classroom_id,
start_time=start_time,
end_time=end_time,
teacher_id=reservation.user_id,
commission_id=reservation.commission_id,
expected_attendees=expected_attendees,
allow_co_teaching_exception=getattr(dto, 'allow_co_teaching_exception', False),
exclude_id=reservation.id
)
if matrix['is_blocked']:
raise ValueError(" | ".join(matrix['block_reasons']))
if dto.classroom_id is not None:
reservation.classroom_id = dto.classroom_id
backend/app/services/sheets_importer.py
+3 -1
View File
@@ -76,11 +76,13 @@ class GoogleSheetsImporter:
"""Fetch CSV string for a specific sheet gid"""
separator = '&' if '?' in self.base_url else '?'
url = f"{self.base_url}{separator}gid={gid}"
if not (url.startswith('https://') or url.startswith('http://')):
raise ValueError(f"URL scheme not permitted: {url}")
req = urllib.request.Request(
url,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) EduSpace/2.0'}
)
with urllib.request.urlopen(req, timeout=25) as resp:
with urllib.request.urlopen(req, timeout=25) as resp: # nosec B310
return resp.read().decode('utf-8', errors='replace')
def parse_sheet_rows(self, csv_content, sheet_name):
backend/migrate_phase2_fields.py
+44
View File
@@ -0,0 +1,44 @@
"""
Migration script for Phase 2 MVP fields:
- subjects.is_short_course (BOOLEAN DEFAULT 0)
- student_enrollments.allow_same_day_exception (BOOLEAN DEFAULT 0)
- student_enrollments.exception_reason (VARCHAR(255))
"""
import sqlite3
import os
def migrate():
db_paths = [
os.path.join(os.path.dirname(__file__), 'instance', 'app.db'),
os.path.join(os.path.dirname(__file__), 'app.db')
]
for db_path in db_paths:
if not os.path.exists(db_path):
continue
print(f"Migrating SQLite DB: {db_path}")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Check subjects.is_short_course
cursor.execute("PRAGMA table_info(subjects)")
columns = [row[1] for row in cursor.fetchall()]
if 'is_short_course' not in columns:
print("Adding column 'is_short_course' to subjects table...")
cursor.execute("ALTER TABLE subjects ADD COLUMN is_short_course BOOLEAN DEFAULT 0")
# Check student_enrollments.allow_same_day_exception
cursor.execute("PRAGMA table_info(student_enrollments)")
enr_columns = [row[1] for row in cursor.fetchall()]
if 'allow_same_day_exception' not in enr_columns:
print("Adding column 'allow_same_day_exception' to student_enrollments table...")
cursor.execute("ALTER TABLE student_enrollments ADD COLUMN allow_same_day_exception BOOLEAN DEFAULT 0")
if 'exception_reason' not in enr_columns:
print("Adding column 'exception_reason' to student_enrollments table...")
cursor.execute("ALTER TABLE student_enrollments ADD COLUMN exception_reason VARCHAR(255)")
conn.commit()
conn.close()
print(f"Migration completed for {db_path}")
if __name__ == '__main__':
migrate()
backend/migrate_user_profile_fields.py
+79
View File
@@ -0,0 +1,79 @@
import sqlite3
import os
import sys
# Locate database
possible_paths = [
os.path.join(os.path.dirname(__file__), 'instance', 'edu_space.db'),
os.path.join(os.path.dirname(__file__), 'instance', 'app.db'),
os.path.join(os.path.dirname(__file__), 'edu_space.db'),
os.path.join(os.path.dirname(__file__), 'app.db'),
]
db_path = None
for p in possible_paths:
if os.path.exists(p):
db_path = p
break
if not db_path:
# Check current directory
for root, dirs, files in os.walk(os.path.dirname(__file__)):
for f in files:
if f.endswith('.db'):
db_path = os.path.join(root, f)
break
if db_path:
break
print(f"Connecting to database: {db_path}")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Check existing columns in users table
cursor.execute("PRAGMA table_info(users)")
columns = [row[1] for row in cursor.fetchall()]
print(f"Existing columns in 'users': {columns}")
new_columns = [
('first_name', 'VARCHAR(100)'),
('last_name', 'VARCHAR(100)'),
('phone', 'VARCHAR(50)'),
('address', 'VARCHAR(255)'),
('document_type', 'VARCHAR(20) DEFAULT "DNI"'),
('document_number', 'VARCHAR(30)')
]
for col_name, col_type in new_columns:
if col_name not in columns:
print(f"Adding column '{col_name}'...")
cursor.execute(f"ALTER TABLE users ADD COLUMN {col_name} {col_type}")
else:
print(f"Column '{col_name}' already exists.")
conn.commit()
# Create index on document_number if not exists
try:
cursor.execute("CREATE INDEX IF NOT EXISTS idx_users_document_number ON users(document_number)")
conn.commit()
print("Index on document_number verified.")
except Exception as e:
print(f"Notice on index: {e}")
# Migrate existing users: split name into first_name and last_name if empty
cursor.execute("SELECT id, name, first_name, last_name FROM users")
users = cursor.fetchall()
migrated = 0
for u_id, name, fn, ln in users:
if not fn and name:
parts = name.strip().split()
first_n = parts[0] if parts else ''
last_n = ' '.join(parts[1:]) if len(parts) > 1 else ''
cursor.execute("UPDATE users SET first_name = ?, last_name = ? WHERE id = ?", (first_n, last_n, u_id))
migrated += 1
conn.commit()
print(f"Successfully migrated {migrated} users with first_name and last_name from name.")
conn.close()
print("Migration completed successfully!")
backend/requirements.txt
+5 -5
View File
@@ -1,4 +1,4 @@
Flask==3.0.0
Flask>=3.1.3
Flask-SQLAlchemy==3.1.1
Flask-Login==0.6.3
Flask-WTF==1.2.1
@@ -6,11 +6,11 @@ Flask-Migrate==4.0.5
Flask-CORS==6.0.2
Flask-Babel==4.0.0
WTForms==3.1.0
Werkzeug==3.0.1
Werkzeug>=3.1.6
SQLAlchemy>=2.0.32
python-dotenv==1.0.0
requests==2.31.0
bleach==6.1.0
python-dotenv>=1.2.2
requests>=2.32.4
bleach>=6.4.0
python-dateutil==2.8.2
Pillow>=10.4.0
email_validator>=2.0.0
backend/tests/test_phase2_rules.py
+293
View File
@@ -0,0 +1,293 @@
import unittest
from datetime import datetime, timedelta
from app import create_app, db
from app.models.classroom import Classroom
from app.models.subject import Subject, Commission, CommissionTeacher
from app.models.user import User
from app.models.role import Role
from app.models.reservation import Reservation
from app.models.enrollment import StudentEnrollment
from app.services.reservation_service import ReservationService
from app.services.enrollment_service import EnrollmentService
from app.schemas.reservation_dto import ReservationCreateDTO
class TestPhase2Rules(unittest.TestCase):
"""Pruebas unitarias e integrales para la Fase 2 del Roadmap MVP (UniCABA)."""
def setUp(self):
self.app = create_app()
self.app_context = self.app.app_context()
self.app_context.push()
self.res_service = ReservationService()
# Obtener o crear rol alumno y docente
self.role_alumno = Role.query.filter_by(name='Alumno').first()
if not self.role_alumno:
self.role_alumno = Role(name='Alumno', description='Rol Alumno')
db.session.add(self.role_alumno)
self.role_docente = Role.query.filter_by(name='Docente').first()
if not self.role_docente:
self.role_docente = Role(name='Docente', description='Rol Docente')
db.session.add(self.role_docente)
db.session.commit()
# Crear usuarios para tests
ts = int(datetime.utcnow().timestamp())
self.docente1 = User(
name=f'Profesor Principal {ts}',
email=f'docente1_{ts}@unicaba.edu.ar',
role_id=self.role_docente.id,
role='Docente',
is_active=True
)
self.docente1.set_password('Secret123!')
self.docente2 = User(
name=f'Profesor Adjunto {ts}',
email=f'docente2_{ts}@unicaba.edu.ar',
role_id=self.role_docente.id,
role='Docente',
is_active=True
)
self.docente2.set_password('Secret123!')
self.alumno1 = User(
name=f'Estudiante Test {ts}',
email=f'alumno_{ts}@unicaba.edu.ar',
role_id=self.role_alumno.id,
role='Alumno',
is_active=True
)
self.alumno1.set_password('Secret123!')
# Aulas física y virtual
self.aula_fisica = Classroom(
code=f'A-501-{ts}',
capacity=30,
building='Sede Central',
floor='Piso 5',
is_active=True
)
self.aula_virtual = Classroom(
code=f'VIRTUAL-MEET-{ts}',
capacity=100,
building='Campus Virtual',
floor='Plataforma Digital',
is_active=True
)
# Asignaturas regulares y curso corto
self.materia_regular_a = Subject(
code=f'REG-A-{ts}',
name=f'Algoritmos y Estructuras {ts}',
is_short_course=False,
is_active=True
)
self.materia_regular_b = Subject(
code=f'REG-B-{ts}',
name=f'Sistemas Operativos {ts}',
is_short_course=False,
is_active=True
)
self.curso_corto = Subject(
code=f'TALLER-{ts}',
name=f'Taller de Herramientas Git {ts}',
is_short_course=True,
is_active=True
)
db.session.add_all([
self.docente1, self.docente2, self.alumno1,
self.aula_fisica, self.aula_virtual,
self.materia_regular_a, self.materia_regular_b, self.curso_corto
])
db.session.commit()
# Comisiones
self.comision_a = Commission(
subject_id=self.materia_regular_a.id,
code=f'COM-A-{ts}',
semester='1C',
year=2026,
teacher_id=self.docente1.id,
max_students=40,
current_students=0,
schedule='Lunes 08:00 - 12:00',
active=True
)
self.comision_b = Commission(
subject_id=self.materia_regular_b.id,
code=f'COM-B-{ts}',
semester='1C',
year=2026,
teacher_id=self.docente1.id,
max_students=40,
current_students=0,
schedule='Lunes 14:00 - 18:00',
active=True
)
self.comision_c_corto = Commission(
subject_id=self.curso_corto.id,
code=f'COM-CORTO-{ts}',
semester='1C',
year=2026,
teacher_id=self.docente2.id,
max_students=20,
current_students=0,
schedule='Lunes 18:00 - 20:00',
active=True
)
db.session.add_all([self.comision_a, self.comision_b, self.comision_c_corto])
db.session.commit()
def tearDown(self):
db.session.rollback()
self.app_context.pop()
def test_01_physical_conflict_blocks_double_booking(self):
"""Estadio 2.2: Debe bloquear doble asignación en misma aula física y horario."""
start = datetime(2026, 10, 5, 8, 0)
end = datetime(2026, 10, 5, 12, 0)
# Primera reserva exitosa
dto1 = ReservationCreateDTO(
classroom_id=self.aula_fisica.id,
commission_id=self.comision_a.id,
user_id=self.docente1.id,
start_time=start,
end_time=end,
expected_attendees=25
)
res1 = self.res_service.create_reservation(dto1)
self.assertIsNotNone(res1.id)
# Segunda reserva en la misma aula solapándose (09:00 a 11:00)
dto2 = ReservationCreateDTO(
classroom_id=self.aula_fisica.id,
commission_id=self.comision_b.id,
user_id=self.docente2.id,
start_time=start + timedelta(hours=1),
end_time=end - timedelta(hours=1),
expected_attendees=20
)
with self.assertRaises(ValueError) as ctx:
self.res_service.create_reservation(dto2)
self.assertIn("Conflicto de horario", str(ctx.exception))
def test_02_capacity_aforo_validation(self):
"""Estadio 2.2: Debe bloquear reserva si asistentes esperados superan capacidad física."""
start = datetime(2026, 10, 6, 8, 0)
end = datetime(2026, 10, 6, 12, 0)
dto = ReservationCreateDTO(
classroom_id=self.aula_fisica.id,
commission_id=self.comision_a.id,
user_id=self.docente1.id,
start_time=start,
end_time=end,
expected_attendees=self.aula_fisica.capacity + 15 # supera capacidad (30)
)
with self.assertRaises(ValueError) as ctx:
self.res_service.create_reservation(dto)
self.assertIn("Conflicto de aforo", str(ctx.exception))
def test_03_teacher_conflict_and_co_teaching_exception(self):
"""Estadio 2.2: Detección de conflicto docente y excepción justificada por co-docencia."""
start = datetime(2026, 10, 7, 14, 0)
end = datetime(2026, 10, 7, 18, 0)
# Docente 1 dicta clase en aula física
dto1 = ReservationCreateDTO(
classroom_id=self.aula_fisica.id,
commission_id=self.comision_a.id,
user_id=self.docente1.id,
start_time=start,
end_time=end,
expected_attendees=20
)
self.res_service.create_reservation(dto1)
# Mismo docente 1 intenta otra clase en sala virtual al mismo tiempo
dto_conflict = ReservationCreateDTO(
classroom_id=self.aula_virtual.id,
commission_id=self.comision_b.id,
user_id=self.docente1.id,
start_time=start,
end_time=end,
expected_attendees=20
)
# Sin co-docencia debe bloquear
with self.assertRaises(ValueError) as ctx:
self.res_service.create_reservation(dto_conflict)
self.assertIn("Conflicto de horario docente", str(ctx.exception))
# Ahora incorporamos a Docente 2 como co-docente en comision_b
ct = CommissionTeacher(
commission_id=self.comision_b.id,
user_id=self.docente2.id,
role='Adjunto'
)
db.session.add(ct)
db.session.commit()
# Con co-docencia registrada, la matriz detecta cobertura docente y permite la reserva
res_co = self.res_service.create_reservation(dto_conflict)
self.assertIsNotNone(res_co.id)
def test_04_student_same_day_restriction_rule(self):
"""Estadio 2.3: Un alumno no puede cursar dos asignaturas regulares el mismo día sin excepción."""
# Inscribir al alumno en comision_a (Lunes regular)
enr1 = EnrollmentService.enroll_student(
commission_id=self.comision_a.id,
student_id=self.alumno1.id
)
self.assertEqual(enr1.status, 'activo')
# Intentar inscribir en comision_b (también Lunes regular): debe BLOQUEAR
with self.assertRaises(ValueError) as ctx:
EnrollmentService.enroll_student(
commission_id=self.comision_b.id,
student_id=self.alumno1.id
)
self.assertIn("Restricción de cursada diaria", str(ctx.exception))
self.assertIn("Lunes", str(ctx.exception))
def test_05_student_short_course_bypass(self):
"""Estadio 2.3: Si una de las materias es curso corto, se autoriza automáticamente."""
# El alumno cursa comision_a los Lunes
EnrollmentService.enroll_student(
commission_id=self.comision_a.id,
student_id=self.alumno1.id
)
# Intentar inscribir en curso corto (Lunes taller): debe PERMITIR
enr_short = EnrollmentService.enroll_student(
commission_id=self.comision_c_corto.id,
student_id=self.alumno1.id
)
self.assertIsNotNone(enr_short.id)
self.assertEqual(enr_short.status, 'activo')
def test_06_student_bedelia_manual_exception_bypass(self):
"""Estadio 2.3: Con autorización expresa de Bedelía (allow_same_day_exception), se permite."""
# Alumno cursa comision_a los Lunes
EnrollmentService.enroll_student(
commission_id=self.comision_a.id,
student_id=self.alumno1.id
)
# Inscribir en comision_b con allow_same_day_exception=True
enr2 = EnrollmentService.enroll_student(
commission_id=self.comision_b.id,
student_id=self.alumno1.id,
allow_same_day_exception=True,
exception_reason="Autorización especial de Bedelía por cambio de plan de estudios"
)
self.assertIsNotNone(enr2.id)
self.assertTrue(enr2.allow_same_day_exception)
self.assertIn("cambio de plan", enr2.exception_reason)
if __name__ == '__main__':
unittest.main()