""" Integration and Unit Tests for Phase 6 (Auth, Email & Moodle Backend) Tests: - CryptoService encryption and decryption - SystemSetting encrypted values and masking - MoodleQueueService (fault-tolerant queue, backoff, Dead Letter Queue DLQ) - CacheService hybrid operation - Hybrid Auth routes (providers, Google OAuth with domain checks, Moodle delegated auth) - EmailService dynamic configuration """ import pytest from datetime import datetime, timedelta from unittest.mock import patch, MagicMock from app.services.crypto_service import crypto_service from app.services.cache_service import cache_service from app.services.moodle_queue_service import moodle_queue_service from app.services.email_service import email_service from app.models.setting import SystemSetting from app.models.sync_task import MoodleSyncTask from app.models.user import User from app.models.role import Role def test_crypto_service_encryption_and_decryption(app_context): secret = "SuperSecretPassword123!" encrypted = crypto_service.encrypt(secret) assert encrypted != secret assert len(encrypted) > 20 decrypted = crypto_service.decrypt(encrypted) assert decrypted == secret def test_system_setting_encrypted_value(init_database, app_context): key = "test_smtp_pass" val = "MySmtpSecretPass2026*" setting = SystemSetting.set_encrypted_value(key, val, "Test SMTP Password", category="smtp") assert setting.is_encrypted is True assert setting.value != val # Must be encrypted in DB assert setting.get_decrypted_value() == val assert setting.get_masked_value() == "••••••••••••" def test_cache_service_fallback(app_context): key = "test_unit_key" val = {"foo": "bar", "num": 42} cache_service.set(key, val, ttl_seconds=10) retrieved = cache_service.get(key) assert retrieved == val cache_service.delete(key) assert cache_service.get(key) is None def test_moodle_queue_service_enqueue_and_dlq(init_database, app_context): # 1. Enqueue task task = moodle_queue_service.enqueue_task( action="CREATE_USER", entity_type="user", entity_id=99, payload={"username": "testuser", "email": "test@unicaba.edu.ar"}, max_attempts=2 ) assert task.status == 'PENDING' assert task.attempts == 0 # 2. Simulate processing failure (attempt 1 -> RETRYING) with patch('app.services.moodle_queue_service.MoodleQueueService._execute_task_action') as mock_exec: mock_exec.side_effect = Exception("Moodle Server Connection Timeout (504)") results1 = moodle_queue_service.process_pending_tasks(batch_size=10) assert results1['processed'] == 1 assert results1['retrying'] == 1 assert results1['failed'] == 0 # Verify task is now RETRYING with exponential backoff next_retry_at task_refreshed = MoodleSyncTask.query.get(task.id) assert task_refreshed.status == 'RETRYING' assert task_refreshed.attempts == 1 assert "504" in task_refreshed.error_message assert task_refreshed.next_retry_at is not None # 3. Simulate second failure with next_retry_at in the past -> moves to DLQ (FAILED) task_refreshed.next_retry_at = datetime.utcnow() - timedelta(minutes=1) init_database.session.commit() results2 = moodle_queue_service.process_pending_tasks(batch_size=10) assert results2['processed'] == 1 assert results2['failed'] == 1 # DLQ triggered task_dlq = MoodleSyncTask.query.get(task.id) assert task_dlq.status == 'FAILED' # 4. Manual DLQ retry res_retry = moodle_queue_service.retry_task(task.id) assert res_retry is True task_reset = MoodleSyncTask.query.get(task.id) assert task_reset.status == 'PENDING' assert task_reset.attempts == 0 def test_email_service_dynamic_config(init_database, app_context): SystemSetting.set_value('smtp_host', 'mail.unicaba.edu.ar', 'Host', category='smtp') SystemSetting.set_value('smtp_port', '465', 'Port', category='smtp') SystemSetting.set_value('smtp_security', 'ssl', 'Sec', category='smtp') SystemSetting.set_value('smtp_user', 'bedelia@unicaba.edu.ar', 'User', category='smtp') SystemSetting.set_encrypted_value('smtp_password', 'BedeliaSecure2026!', 'Pass', category='smtp') cfg = email_service.get_smtp_config() assert cfg['host'] == 'mail.unicaba.edu.ar' assert cfg['port'] == 465 assert cfg['security'] == 'ssl' assert cfg['user'] == 'bedelia@unicaba.edu.ar' assert cfg['password'] == 'BedeliaSecure2026!' def test_auth_providers_public_endpoint(client, init_database): SystemSetting.set_value('auth_local_enabled', 'true') SystemSetting.set_value('auth_google_enabled', 'true') SystemSetting.set_value('auth_moodle_enabled', 'false') SystemSetting.set_value('google_client_id', 'google-test-id-123') res = client.get('/api/v1/auth/providers') assert res.status_code == 200 data = res.get_json() assert data['status'] == 'success' assert data['providers']['local'] is True assert data['providers']['google'] is True assert data['providers']['moodle'] is False assert data['google_client_id'] == 'google-test-id-123' def test_google_auth_domain_enforcement(client, init_database): SystemSetting.set_value('auth_google_enabled', 'true') SystemSetting.set_value('google_allowed_domains', 'unicaba.edu.ar,lasalle.edu.ar') # Domain rejected (e.g., hacker@gmail.com) res_bad = client.post('/api/v1/auth/google', json={ 'email': 'hacker@gmail.com', 'name': 'Hacker Unauthorized', 'domain': 'gmail.com' }) assert res_bad.status_code == 403 assert "no está autorizado" in res_bad.get_json()['message'] # Domain accepted (docente@unicaba.edu.ar) res_good = client.post('/api/v1/auth/google', json={ 'email': 'docente.nuevo@unicaba.edu.ar', 'name': 'Docente Nuevo', 'domain': 'unicaba.edu.ar' }) assert res_good.status_code == 200 data_good = res_good.get_json() assert 'access_token' in data_good assert data_good['user']['email'] == 'docente.nuevo@unicaba.edu.ar' assert data_good['user']['role'] == 'Docente'