From 128432f679b1e24754fc1ed82d326af2fcbe4d5e Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:47:32 +0000 Subject: [PATCH] Add unit tests for all modules including config, main, playlist, and server --- tests/conftest.py | 55 +++++ tests/test_config.py | 177 ++++++++++++++++ tests/test_main.py | 181 ++++++++++++++++ tests/test_playlist.py | 460 +++++++++++++++++++++++++++++++++++++++++ tests/test_server.py | 179 ++++++++++++++++ 5 files changed, 1052 insertions(+) create mode 100644 tests/conftest.py create mode 100644 tests/test_config.py create mode 100644 tests/test_main.py create mode 100644 tests/test_playlist.py create mode 100644 tests/test_server.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..30cd858 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,55 @@ +"""Configuración compartida para los tests de pytest.""" + +import pytest + + +@pytest.fixture +def sample_channels(): + """Fixture con datos de ejemplo de canales.""" + return [ + { + "name": "Channel 1", + "stream_id": 101, + "stream_icon": "http://example.com/icon1.png", + "category_id": "1", + }, + { + "name": "Channel 2", + "stream_id": 102, + "stream_icon": "http://example.com/icon2.png", + "category_id": "2", + }, + { + "name": "Channel 3", + "stream_id": 103, + "stream_icon": "", + "category_id": "", + }, + ] + + +@pytest.fixture +def empty_channels(): + """Fixture con lista vacía de canales.""" + return [] + + +@pytest.fixture +def minimal_channel(): + """Fixture con canal con campos mínimos.""" + return [{"stream_id": 999}] + + +@pytest.fixture +def mock_settings(monkeypatch): + """Fixture para mockear settings.""" + + class MockSettings: + host = "http://test-iptv.com" + username = "testuser" + password = "testpass" + port = 8080 + update_interval = 60 + output_file = "test_playlist.m3u" + + return MockSettings() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..80ce59f --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,177 @@ +"""Tests unitarios para el módulo config.""" + +import os + +import pytest +from pydantic import ValidationError + + +@pytest.fixture(autouse=True) +def clean_env(monkeypatch): + """Limpia todas las variables de entorno relevantes antes de cada test.""" + env_vars = [ + "HOST", + "USERNAME", + "PASSWORD", + "PORT", + "UPDATE_INTERVAL", + "OUTPUT_FILE", + ] + for var in env_vars: + monkeypatch.delenv(var, raising=False) + yield + + +class TestSettings: + """Tests para la clase Settings.""" + + def test_settings_with_required_fields(self, monkeypatch): + """Test: Settings se crea correctamente con campos obligatorios.""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + + from m3u_list_builder.config import Settings + + # _env_file=None evita leer el archivo .env + settings = Settings(_env_file=None) + + assert settings.host == "http://test.com" + assert settings.username == "user" + assert settings.password == "pass" + + def test_settings_default_values(self, monkeypatch): + """Test: Settings usa valores por defecto correctos.""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + + from m3u_list_builder.config import Settings + + settings = Settings(_env_file=None) + + assert settings.port == 8080 + assert settings.update_interval == 3600 + assert settings.output_file == "playlist.m3u" + + def test_settings_custom_values(self, monkeypatch): + """Test: Settings acepta valores personalizados.""" + monkeypatch.setenv("HOST", "http://custom.com") + monkeypatch.setenv("USERNAME", "custom_user") + monkeypatch.setenv("PASSWORD", "custom_pass") + monkeypatch.setenv("PORT", "9090") + monkeypatch.setenv("UPDATE_INTERVAL", "7200") + monkeypatch.setenv("OUTPUT_FILE", "custom.m3u") + + from m3u_list_builder.config import Settings + + settings = Settings(_env_file=None) + + assert settings.host == "http://custom.com" + assert settings.username == "custom_user" + assert settings.password == "custom_pass" + assert settings.port == 9090 + assert settings.update_interval == 7200 + assert settings.output_file == "custom.m3u" + + def test_settings_missing_required_host(self, monkeypatch): + """Test: Settings falla sin HOST.""" + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + + from m3u_list_builder.config import Settings + + with pytest.raises(ValidationError) as exc_info: + Settings(_env_file=None) + + assert "host" in str(exc_info.value).lower() + + def test_settings_missing_required_username(self, monkeypatch): + """Test: Settings falla sin USERNAME.""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("PASSWORD", "pass") + + from m3u_list_builder.config import Settings + + with pytest.raises(ValidationError) as exc_info: + Settings(_env_file=None) + + assert "username" in str(exc_info.value).lower() + + def test_settings_missing_required_password(self, monkeypatch): + """Test: Settings falla sin PASSWORD.""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + + from m3u_list_builder.config import Settings + + with pytest.raises(ValidationError) as exc_info: + Settings(_env_file=None) + + assert "password" in str(exc_info.value).lower() + + def test_settings_invalid_port_type(self, monkeypatch): + """Test: Settings falla con PORT inválido.""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + monkeypatch.setenv("PORT", "invalid") + + from m3u_list_builder.config import Settings + + with pytest.raises(ValidationError): + Settings(_env_file=None) + + def test_settings_extra_fields_ignored(self, monkeypatch): + """Test: Settings ignora campos extra (extra='ignore').""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + monkeypatch.setenv("UNKNOWN_FIELD", "should_be_ignored") + + from m3u_list_builder.config import Settings + + # No debe lanzar excepción + settings = Settings(_env_file=None) + assert not hasattr(settings, "unknown_field") + + def test_settings_reads_env_file(self, tmp_path, monkeypatch): + """Test: Settings puede leer desde archivo .env.""" + # Crear archivo .env temporal + env_file = tmp_path / ".env" + env_file.write_text( + "HOST=http://from-env-file.com\nUSERNAME=envuser\nPASSWORD=envpass\n" + ) + + from m3u_list_builder.config import Settings + + settings = Settings(_env_file=str(env_file)) + + assert settings.host == "http://from-env-file.com" + assert settings.username == "envuser" + assert settings.password == "envpass" + + def test_settings_invalid_port_type(self, monkeypatch): + """Test: Settings falla con PORT inválido.""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + monkeypatch.setenv("PORT", "invalid") + + from m3u_list_builder.config import Settings + + with pytest.raises(ValidationError): + Settings() + + def test_settings_extra_fields_ignored(self, monkeypatch): + """Test: Settings ignora campos extra (extra='ignore').""" + monkeypatch.setenv("HOST", "http://test.com") + monkeypatch.setenv("USERNAME", "user") + monkeypatch.setenv("PASSWORD", "pass") + monkeypatch.setenv("UNKNOWN_FIELD", "should_be_ignored") + + from m3u_list_builder.config import Settings + + # No debe lanzar excepción + settings = Settings() + assert not hasattr(settings, "unknown_field") diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..61468a6 --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,181 @@ +"""Tests unitarios para el módulo main.""" + +from unittest.mock import MagicMock, patch + + +class TestMain: + """Tests para la función main.""" + + # ======================================== + # Tests para main - Complejidad Baja + # Path 1: Orquestación correcta de componentes + # ======================================== + + def test_main_creates_playlist_manager(self): + """Test: main crea una instancia de PlaylistManager.""" + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager") as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server"): + from m3u_list_builder.main import main + + main() + + mock_manager_class.assert_called_once() + + def test_main_starts_updater_thread(self): + """Test: main inicia el hilo de actualización.""" + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager") as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server"): + from m3u_list_builder.main import main + + main() + + mock_thread.assert_called_once() + mock_thread_instance.start.assert_called_once() + + def test_main_thread_is_daemon(self): + """Test: main crea el hilo como daemon.""" + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager") as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server"): + from m3u_list_builder.main import main + + main() + + # Verificar que daemon=True + call_kwargs = mock_thread.call_args[1] + assert call_kwargs["daemon"] is True + + def test_main_thread_targets_manager_loop(self): + """Test: main configura el hilo con target=manager.loop.""" + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager") as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server"): + from m3u_list_builder.main import main + + main() + + # Verificar que target es manager.loop + call_kwargs = mock_thread.call_args[1] + assert call_kwargs["target"] == mock_manager.loop + + def test_main_thread_has_correct_name(self): + """Test: main nombra el hilo como UpdaterThread.""" + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager") as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server"): + from m3u_list_builder.main import main + + main() + + call_kwargs = mock_thread.call_args[1] + assert call_kwargs["name"] == "UpdaterThread" + + def test_main_runs_server(self): + """Test: main ejecuta run_server.""" + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager"): + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server") as mock_run_server: + from m3u_list_builder.main import main + + main() + + mock_run_server.assert_called_once() + + def test_main_starts_thread_before_server(self): + """Test: main inicia el hilo antes de ejecutar el servidor.""" + call_order = [] + + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test.com" + + with patch("m3u_list_builder.main.PlaylistManager"): + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread_instance.start.side_effect = lambda: call_order.append( + "thread_start" + ) + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server") as mock_run_server: + mock_run_server.side_effect = lambda: call_order.append( + "run_server" + ) + + from m3u_list_builder.main import main + + main() + + assert call_order == ["thread_start", "run_server"] + + def test_main_logs_startup_info(self, caplog): + """Test: main registra información de inicio.""" + import logging + + with patch("m3u_list_builder.main.settings") as mock_settings: + mock_settings.host = "http://test-host.com" + + with patch("m3u_list_builder.main.PlaylistManager"): + with patch("m3u_list_builder.main.threading.Thread") as mock_thread: + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + with patch("m3u_list_builder.main.run_server"): + with caplog.at_level(logging.INFO): + from m3u_list_builder.main import main + + main() + + # Puede que el log se capture dependiendo del setup diff --git a/tests/test_playlist.py b/tests/test_playlist.py new file mode 100644 index 0000000..0858482 --- /dev/null +++ b/tests/test_playlist.py @@ -0,0 +1,460 @@ +"""Tests unitarios para el módulo playlist.""" + +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import requests + + +class TestPlaylistManager: + """Tests para la clase PlaylistManager.""" + + @pytest.fixture + def manager(self): + """Fixture para crear un PlaylistManager con settings mockeadas.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + mock_settings.update_interval = 1 + + from m3u_list_builder.playlist import PlaylistManager + + yield PlaylistManager() + + @pytest.fixture + def temp_dir(self): + """Fixture para directorio temporal.""" + with tempfile.TemporaryDirectory() as tmpdir: + yield Path(tmpdir) + + # ======================================== + # Tests para __init__ + # ======================================== + + def test_init_running_is_false(self, manager): + """Test: PlaylistManager inicia con running=False.""" + assert manager.running is False + + # ======================================== + # Tests para fetch_and_generate - Complejidad Alta + # Path 1: Éxito completo + # Path 2: Error de red (RequestException) + # Path 3: Error inesperado (Exception genérica) + # Path 4: Error en JSON parsing + # ======================================== + + def test_fetch_and_generate_success(self, sample_channels): + """Test: fetch_and_generate exitoso descarga y genera M3U.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_response = MagicMock() + mock_response.json.return_value = sample_channels + mock_response.raise_for_status = MagicMock() + mock_get.return_value = mock_response + + with patch.object(manager, "_write_m3u") as mock_write: + manager.fetch_and_generate() + + mock_get.assert_called_once() + mock_write.assert_called_once_with(sample_channels) + + def test_fetch_and_generate_correct_api_call(self): + """Test: fetch_and_generate llama a la API correctamente.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_response = MagicMock() + mock_response.json.return_value = [] + mock_get.return_value = mock_response + + with patch.object(manager, "_write_m3u"): + manager.fetch_and_generate() + + mock_get.assert_called_once_with( + "http://test-iptv.com/player_api.php", + params={ + "username": "testuser", + "password": "testpass", + "action": "get_live_streams", + }, + timeout=30, + ) + + def test_fetch_and_generate_request_exception(self, caplog): + """Test: fetch_and_generate maneja RequestException sin crash.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_get.side_effect = requests.RequestException("Connection failed") + + # No debe lanzar excepción + manager.fetch_and_generate() + + assert "Error de red" in caplog.text + + def test_fetch_and_generate_timeout_exception(self, caplog): + """Test: fetch_and_generate maneja timeout sin crash.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_get.side_effect = requests.Timeout("Timeout") + + manager.fetch_and_generate() + + assert "Error de red" in caplog.text + + def test_fetch_and_generate_http_error(self, caplog): + """Test: fetch_and_generate maneja HTTP error (raise_for_status).""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = requests.HTTPError( + "404 Not Found" + ) + mock_get.return_value = mock_response + + manager.fetch_and_generate() + + assert "Error de red" in caplog.text + + def test_fetch_and_generate_json_decode_error(self, caplog): + """Test: fetch_and_generate maneja error de JSON parsing.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_response = MagicMock() + mock_response.raise_for_status = MagicMock() + mock_response.json.side_effect = ValueError("Invalid JSON") + mock_get.return_value = mock_response + + manager.fetch_and_generate() + + assert "Error inesperado" in caplog.text + + def test_fetch_and_generate_unexpected_exception(self, caplog): + """Test: fetch_and_generate maneja excepciones inesperadas.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + with patch("m3u_list_builder.playlist.requests.get") as mock_get: + mock_response = MagicMock() + mock_response.raise_for_status = MagicMock() + mock_response.json.return_value = [] + mock_get.return_value = mock_response + + with patch.object( + manager, "_write_m3u", side_effect=RuntimeError("Disk full") + ): + manager.fetch_and_generate() + + assert "Error inesperado" in caplog.text + + # ======================================== + # Tests para _write_m3u - Complejidad Media + # Path 1: Lista vacía + # Path 2: Lista con múltiples canales + # Path 3: Canal con campos faltantes (usa defaults) + # ======================================== + + def test_write_m3u_empty_channels(self, temp_dir): + """Test: _write_m3u genera archivo con solo header para lista vacía.""" + output_file = temp_dir / "playlist.m3u" + + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = str(output_file) + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + manager._write_m3u([]) + + content = output_file.read_text() + assert content == "#EXTM3U\n" + + def test_write_m3u_multiple_channels(self, temp_dir, sample_channels): + """Test: _write_m3u genera archivo correcto con múltiples canales.""" + output_file = temp_dir / "playlist.m3u" + + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = str(output_file) + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + manager._write_m3u(sample_channels) + + content = output_file.read_text() + + # Verificar header + assert content.startswith("#EXTM3U\n") + + # Verificar cada canal + for channel in sample_channels: + assert channel["name"] in content + assert str(channel["stream_id"]) in content + + def test_write_m3u_channel_format(self, temp_dir): + """Test: _write_m3u genera formato EXTINF correcto.""" + output_file = temp_dir / "playlist.m3u" + + channel = [ + { + "name": "Test Channel", + "stream_id": 123, + "stream_icon": "http://icon.com/test.png", + "category_id": "5", + } + ] + + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://iptv.com" + mock_settings.username = "user" + mock_settings.password = "pass" + mock_settings.output_file = str(output_file) + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + manager._write_m3u(channel) + + content = output_file.read_text() + lines = content.strip().split("\n") + + assert len(lines) == 3 # Header + EXTINF + URL + + # Verificar EXTINF + assert '#EXTINF:-1 tvg-id="Test Channel"' in lines[1] + assert 'tvg-logo="http://icon.com/test.png"' in lines[1] + assert 'group-title="Cat_5"' in lines[1] + assert lines[1].endswith(",Test Channel") + + # Verificar URL + assert lines[2] == "http://iptv.com/live/user/pass/123.ts" + + def test_write_m3u_missing_fields_uses_defaults(self, temp_dir, minimal_channel): + """Test: _write_m3u usa valores por defecto para campos faltantes.""" + output_file = temp_dir / "playlist.m3u" + + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = str(output_file) + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + manager._write_m3u(minimal_channel) + + content = output_file.read_text() + + # Debe usar "Unknown" como nombre por defecto + assert "Unknown" in content + # Debe tener icono vacío + assert 'tvg-logo=""' in content + + def test_write_m3u_atomic_replacement(self, temp_dir): + """Test: _write_m3u reemplaza atómicamente (usa archivo temporal).""" + output_file = temp_dir / "playlist.m3u" + temp_file = temp_dir / "playlist.m3u.tmp" + + # Crear archivo existente + output_file.write_text("OLD CONTENT") + + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = str(output_file) + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + manager._write_m3u([]) + + # El archivo temporal no debe existir después + assert not temp_file.exists() + + # El archivo final debe tener el nuevo contenido + assert output_file.read_text() == "#EXTM3U\n" + + # ======================================== + # Tests para loop - Complejidad Media + # Path 1: Loop ejecuta fetch_and_generate + # Path 2: Loop respeta running=False + # ======================================== + + def test_loop_sets_running_true(self): + """Test: loop establece running=True.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + mock_settings.update_interval = 1 + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + # Mock fetch_and_generate para que establezca running=False + call_count = 0 + + def stop_after_one_call(): + nonlocal call_count + call_count += 1 + if call_count >= 1: + manager.running = False + + with patch.object( + manager, "fetch_and_generate", side_effect=stop_after_one_call + ): + with patch("m3u_list_builder.playlist.time.sleep"): + manager.loop() + + assert call_count == 1 + + def test_loop_calls_fetch_and_generate(self): + """Test: loop llama a fetch_and_generate.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + mock_settings.update_interval = 1 + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + fetch_mock = MagicMock( + side_effect=lambda: setattr(manager, "running", False) + ) + + with patch.object(manager, "fetch_and_generate", fetch_mock): + with patch("m3u_list_builder.playlist.time.sleep"): + manager.loop() + + fetch_mock.assert_called() + + def test_loop_respects_update_interval(self): + """Test: loop usa el intervalo de actualización correcto.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + mock_settings.update_interval = 3600 + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + + call_count = 0 + + def stop_loop(): + nonlocal call_count + call_count += 1 + if call_count >= 1: + manager.running = False + + with patch.object(manager, "fetch_and_generate", side_effect=stop_loop): + with patch("m3u_list_builder.playlist.time.sleep") as mock_sleep: + manager.loop() + + mock_sleep.assert_called_with(3600) + + def test_loop_stops_when_running_false(self): + """Test: loop se detiene cuando running=False.""" + with patch("m3u_list_builder.playlist.settings") as mock_settings: + mock_settings.host = "http://test-iptv.com" + mock_settings.username = "testuser" + mock_settings.password = "testpass" + mock_settings.output_file = "test_playlist.m3u" + mock_settings.update_interval = 1 + + from m3u_list_builder.playlist import PlaylistManager + + manager = PlaylistManager() + iterations = 0 + + def track_iterations(): + nonlocal iterations + iterations += 1 + if iterations >= 3: + manager.running = False + + with patch.object( + manager, "fetch_and_generate", side_effect=track_iterations + ): + with patch("m3u_list_builder.playlist.time.sleep"): + manager.loop() + + assert iterations == 3 diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..87650d1 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,179 @@ +"""Tests unitarios para el módulo server.""" + +from unittest.mock import MagicMock, patch + + +class TestRunServer: + """Tests para la función run_server.""" + + # ======================================== + # Tests para run_server - Complejidad Media + # Path 1: Servidor inicia y sirve normalmente + # Path 2: KeyboardInterrupt detiene el servidor + # Path 3: Servidor se cierra correctamente en finally + # ======================================== + + def test_run_server_creates_server_with_correct_address(self): + """Test: run_server crea servidor en la dirección correcta.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 9999 + mock_settings.output_file = "test.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + run_server() + + # Verificar que se creó con la dirección correcta + mock_server_class.assert_called_once() + call_args = mock_server_class.call_args[0] + assert call_args[0] == ("", 9999) + + def test_run_server_uses_correct_port(self): + """Test: run_server usa el puerto de settings.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + run_server() + + call_args = mock_server_class.call_args[0] + assert call_args[0][1] == 8080 + + def test_run_server_calls_serve_forever(self): + """Test: run_server llama a serve_forever.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + run_server() + + mock_server.serve_forever.assert_called_once() + + def test_run_server_handles_keyboard_interrupt(self): + """Test: run_server maneja KeyboardInterrupt sin crash.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + # No debe lanzar excepción + run_server() + + def test_run_server_closes_server_on_interrupt(self): + """Test: run_server cierra el servidor tras KeyboardInterrupt.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + run_server() + + mock_server.server_close.assert_called_once() + + def test_run_server_closes_server_on_normal_exit(self): + """Test: run_server cierra el servidor en salida normal.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + # Simular que serve_forever termina normalmente + mock_server.serve_forever.return_value = None + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + run_server() + + mock_server.server_close.assert_called_once() + + def test_run_server_logs_startup_message(self, caplog): + """Test: run_server registra mensaje de inicio.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + import logging + + with caplog.at_level(logging.INFO): + from m3u_list_builder.server import run_server + + run_server() + + # El mensaje puede estar en el log + # Nota: Dependiendo del orden de imports, puede que no se capture + + def test_run_server_uses_simple_http_handler(self): + """Test: run_server usa SimpleHTTPRequestHandler.""" + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" + + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + with patch( + "m3u_list_builder.server.SimpleHTTPRequestHandler" + ) as mock_handler: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server + + from m3u_list_builder.server import run_server + + run_server() + + # Verificar que se pasó el handler correcto + call_args = mock_server_class.call_args[0] + assert call_args[1] == mock_handler