From 7de3aa5b94250df17b4470fcaba8b1372c18ad3c Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:28:09 +0000 Subject: [PATCH 01/15] Update project name and description in pyproject.toml --- pyproject.toml | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index f87a622..2cd7543 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,10 +1,10 @@ [tool.poetry] -name = "my-project" +name = "m3u_list_builder" version = "0.0.0" -description = "Python boilerplate for Gitea with Docker and Poetry" +description = "Python tool to build M3U lists from various sources." authors = ["Unai Blazquez "] readme = "README.md" -packages = [{include = "my_project", from = "src"}] +packages = [{ include = "m3u_list_builder", from = "src" }] [tool.poetry.dependencies] python = "^3.14" @@ -28,10 +28,13 @@ target-version = "py314" [tool.ruff.lint] # E/F: Errores base, I: Imports (isort), D: Docstrings select = ["E", "F", "I", "D"] -ignore = ["D100", "D104"] # Ignorar docstring en modulos/paquetes vacíos si se desea +ignore = [ + "D100", + "D104", +] # Ignorar docstring en modulos/paquetes vacíos si se desea [tool.ruff.lint.pydocstyle] -convention = "google" # Estilo de docstring (Google, NumPy o PEP 257) +convention = "google" # Estilo de docstring (Google, NumPy o PEP 257) # --- Configuración de Coverage --- [tool.coverage.run] @@ -39,5 +42,5 @@ source = ["src"] branch = true [tool.coverage.report] -fail_under = 80 # CI falla si el coverage es menor al 80% +fail_under = 80 # CI falla si el coverage es menor al 80% show_missing = true From d9dfe7b7c415fa9eba455882dab98c093dbfa6dc Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:33:15 +0000 Subject: [PATCH 02/15] Update CI/CD pipeline to trigger on all branches --- .gitea/workflows/ci.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 0c6e37a..ec22008 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -2,7 +2,7 @@ name: CI/CD Pipeline on: push: - branches: [main] + branches: ['**'] tags: ['v*'] pull_request: From 5342644b1d6855e5e4b8f8fd9f7b931401b30f89 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:36:40 +0000 Subject: [PATCH 03/15] Liniting --- src/m3u_list_builder/main.py | 3 +-- src/m3u_list_builder/playlist.py | 15 ++++++++++----- src/m3u_list_builder/server.py | 2 +- 3 files changed, 12 insertions(+), 8 deletions(-) diff --git a/src/m3u_list_builder/main.py b/src/m3u_list_builder/main.py index a38460e..c716d6f 100644 --- a/src/m3u_list_builder/main.py +++ b/src/m3u_list_builder/main.py @@ -1,5 +1,4 @@ -""" -Módulo principal para iniciar el servicio de construcción de listas M3U.""" +"""Módulo principal para iniciar el servicio de construcción de listas M3U.""" import logging import sys diff --git a/src/m3u_list_builder/playlist.py b/src/m3u_list_builder/playlist.py index c417814..860732d 100644 --- a/src/m3u_list_builder/playlist.py +++ b/src/m3u_list_builder/playlist.py @@ -14,6 +14,7 @@ class PlaylistManager: """Clase para gestionar la generación y actualización de listas M3U.""" def __init__(self): + """Inicialize the PlaylistManager.""" self.running = False def fetch_and_generate(self): @@ -44,7 +45,6 @@ class PlaylistManager: def _write_m3u(self, channels: list): """Escribe el archivo M3U en disco de forma atómica.""" - # Escribimos en un temporal y renombramos para evitar lecturas de archivo corrupto temp_file = Path(f"{settings.output_file}.tmp") final_file = Path(settings.output_file) @@ -57,11 +57,16 @@ class PlaylistManager: cat_id = channel.get("category_id", "") # Construir URL directa - stream_url = f"{settings.host}/live/{settings.username}/{settings.password}/{stream_id}.ts" - - f.write( - f'#EXTINF:-1 tvg-id="{name}" tvg-logo="{icon}" group-title="Cat_{cat_id}",{name}\n' + stream_url = ( + f"{settings.host}/live/{settings.username}/" + f"{settings.password}/{stream_id}.ts" ) + + extinf_line = ( + f'#EXTINF:-1 tvg-id="{name}" tvg-logo="{icon}" ' + f'group-title="Cat_{cat_id}",{name}\n' + ) + f.write(extinf_line) f.write(f"{stream_url}\n") # Reemplazo atómico diff --git a/src/m3u_list_builder/server.py b/src/m3u_list_builder/server.py index 4c97007..a4e75e6 100644 --- a/src/m3u_list_builder/server.py +++ b/src/m3u_list_builder/server.py @@ -9,7 +9,7 @@ logger = logging.getLogger(__name__) def run_server(): """Inicia el servidor HTTP bloqueante.""" - handler = partial(SimpleHTTPRequestHandler, directory=".") + _ = partial(SimpleHTTPRequestHandler, directory=".") # Truco: SimpleHTTPRequestHandler sirve el directorio actual, # asegurarse de que el CWD es correcto o mover el archivo a una carpeta 'public' From 9f7d3d98c1645ed466acffb9f23cc7f1a5dfdf5d Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:47:14 +0000 Subject: [PATCH 04/15] Refactor import statements to use the correct module path --- src/m3u_list_builder/main.py | 6 +++--- src/m3u_list_builder/playlist.py | 3 ++- src/m3u_list_builder/server.py | 2 +- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/m3u_list_builder/main.py b/src/m3u_list_builder/main.py index c716d6f..3e3797c 100644 --- a/src/m3u_list_builder/main.py +++ b/src/m3u_list_builder/main.py @@ -4,9 +4,9 @@ import logging import sys import threading -from my_project.config import settings -from my_project.playlist import PlaylistManager -from my_project.server import run_server +from m3u_list_builder.config import settings +from m3u_list_builder.playlist import PlaylistManager +from m3u_list_builder.server import run_server # Configuración básica de logging logging.basicConfig( diff --git a/src/m3u_list_builder/playlist.py b/src/m3u_list_builder/playlist.py index 860732d..35da940 100644 --- a/src/m3u_list_builder/playlist.py +++ b/src/m3u_list_builder/playlist.py @@ -5,7 +5,8 @@ import time from pathlib import Path import requests -from my_project.config import settings + +from m3u_list_builder.config import settings logger = logging.getLogger(__name__) diff --git a/src/m3u_list_builder/server.py b/src/m3u_list_builder/server.py index a4e75e6..38ac3b4 100644 --- a/src/m3u_list_builder/server.py +++ b/src/m3u_list_builder/server.py @@ -2,7 +2,7 @@ import logging from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer -from my_project.config import settings +from m3u_list_builder.config import settings logger = logging.getLogger(__name__) From 128432f679b1e24754fc1ed82d326af2fcbe4d5e Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:47:32 +0000 Subject: [PATCH 05/15] 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 From 4bf59efd6f35c7d2aa6ac6bba26d50074886b7f9 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:49:02 +0000 Subject: [PATCH 06/15] Update Ruff configuration to exclude tests from linting --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 2cd7543..9870a68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ build-backend = "poetry.core.masonry.api" [tool.ruff] line-length = 88 target-version = "py314" +exclude = ["tests"] [tool.ruff.lint] # E/F: Errores base, I: Imports (isort), D: Docstrings From 0dc6c3607b1c2efff4973e53975572705d4db005 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 18:52:19 +0000 Subject: [PATCH 07/15] Update test coverage source path and adjust pytest command for consistency --- .gitea/workflows/ci.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index ec22008..2e11516 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -27,7 +27,7 @@ jobs: run: poetry run ruff check . - name: Run Tests with Coverage - run: poetry run pytest --cov=my_project --cov-report=term-missing --cov-fail-under=80 + run: poetry run pytest --cov --cov-report=term-missing --cov-fail-under=80 publish-container: needs: test-and-lint From fd20bb28f2af521ded6bca0be14c1ec734604c07 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:12:52 +0000 Subject: [PATCH 08/15] Add public directory to .gitignore for generated files --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index 6448c00..5d2d8e4 100644 --- a/.gitignore +++ b/.gitignore @@ -42,3 +42,6 @@ Thumbs.db # --- Logs --- *.log + +# --- Archivos Generados --- +public/ From cbf22422e3d5b0a80c8db63943f98d4de1fff156 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:13:42 +0000 Subject: [PATCH 09/15] Refactor PlaylistManager to use PUBLIC_DIR for file operations and update tests accordingly --- src/m3u_list_builder/playlist.py | 9 +- tests/test_playlist.py | 153 ++++++++++++++++--------------- 2 files changed, 84 insertions(+), 78 deletions(-) diff --git a/src/m3u_list_builder/playlist.py b/src/m3u_list_builder/playlist.py index 35da940..84ef5f7 100644 --- a/src/m3u_list_builder/playlist.py +++ b/src/m3u_list_builder/playlist.py @@ -10,6 +10,9 @@ from m3u_list_builder.config import settings logger = logging.getLogger(__name__) +# Directorio dedicado para servir archivos +PUBLIC_DIR = Path("public") + class PlaylistManager: """Clase para gestionar la generación y actualización de listas M3U.""" @@ -17,6 +20,8 @@ class PlaylistManager: def __init__(self): """Inicialize the PlaylistManager.""" self.running = False + # Asegurar que el directorio público existe + PUBLIC_DIR.mkdir(exist_ok=True) def fetch_and_generate(self): """Descarga datos y regenera el archivo M3U.""" @@ -46,8 +51,8 @@ class PlaylistManager: def _write_m3u(self, channels: list): """Escribe el archivo M3U en disco de forma atómica.""" - temp_file = Path(f"{settings.output_file}.tmp") - final_file = Path(settings.output_file) + temp_file = PUBLIC_DIR / f"{settings.output_file}.tmp" + final_file = PUBLIC_DIR / settings.output_file with open(temp_file, "w", encoding="utf-8") as f: f.write("#EXTM3U\n") diff --git a/tests/test_playlist.py b/tests/test_playlist.py index 0858482..fb9d39a 100644 --- a/tests/test_playlist.py +++ b/tests/test_playlist.py @@ -219,51 +219,49 @@ class TestPlaylistManager: 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.PUBLIC_DIR", temp_dir): + 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 = "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 - from m3u_list_builder.playlist import PlaylistManager + manager = PlaylistManager() + manager._write_m3u([]) - manager = PlaylistManager() - manager._write_m3u([]) - - content = output_file.read_text() - assert content == "#EXTM3U\n" + output_file = temp_dir / "playlist.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.PUBLIC_DIR", temp_dir): + 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 = "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 - from m3u_list_builder.playlist import PlaylistManager + manager = PlaylistManager() + manager._write_m3u(sample_channels) - manager = PlaylistManager() - manager._write_m3u(sample_channels) + output_file = temp_dir / "playlist.m3u" + content = output_file.read_text() - content = output_file.read_text() + # Verificar header + assert content.startswith("#EXTM3U\n") - # 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 + # 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", @@ -273,52 +271,54 @@ class TestPlaylistManager: } ] - 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) + with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir): + 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 = "playlist.m3u" - from m3u_list_builder.playlist import PlaylistManager + from m3u_list_builder.playlist import PlaylistManager - manager = PlaylistManager() - manager._write_m3u(channel) + manager = PlaylistManager() + manager._write_m3u(channel) - content = output_file.read_text() - lines = content.strip().split("\n") + output_file = temp_dir / "playlist.m3u" + content = output_file.read_text() + lines = content.strip().split("\n") - assert len(lines) == 3 # Header + EXTINF + URL + 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 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" + # 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.PUBLIC_DIR", temp_dir): + 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 = "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 - from m3u_list_builder.playlist import PlaylistManager + manager = PlaylistManager() + manager._write_m3u(minimal_channel) - manager = PlaylistManager() - manager._write_m3u(minimal_channel) + output_file = temp_dir / "playlist.m3u" + content = output_file.read_text() - 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 + # 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).""" @@ -328,22 +328,23 @@ class TestPlaylistManager: # 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) + with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir): + 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 = "playlist.m3u" - from m3u_list_builder.playlist import PlaylistManager + from m3u_list_builder.playlist import PlaylistManager - manager = PlaylistManager() - manager._write_m3u([]) + manager = PlaylistManager() + manager._write_m3u([]) - # El archivo temporal no debe existir después - assert not temp_file.exists() + # 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" + # El archivo final debe tener el nuevo contenido + assert output_file.read_text() == "#EXTM3U\n" # ======================================== # Tests para loop - Complejidad Media From 3c3f2354e303b01b8b55f49e5ffadf89ed10f94e Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:14:23 +0000 Subject: [PATCH 10/15] Refactor server module to use dedicated public directory and enhance tests for run_server and get_public_dir functions --- src/m3u_list_builder/server.py | 19 ++- tests/test_server.py | 287 ++++++++++++++++++++------------- 2 files changed, 190 insertions(+), 116 deletions(-) diff --git a/src/m3u_list_builder/server.py b/src/m3u_list_builder/server.py index 38ac3b4..b969f63 100644 --- a/src/m3u_list_builder/server.py +++ b/src/m3u_list_builder/server.py @@ -1,22 +1,31 @@ import logging from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path from m3u_list_builder.config import settings logger = logging.getLogger(__name__) +# Directorio dedicado para servir archivos (solo playlist) +PUBLIC_DIR = Path("public") + + +def get_public_dir() -> Path: + """Retorna el directorio público, creándolo si no existe.""" + PUBLIC_DIR.mkdir(exist_ok=True) + return PUBLIC_DIR + def run_server(): """Inicia el servidor HTTP bloqueante.""" - _ = partial(SimpleHTTPRequestHandler, directory=".") + public_dir = get_public_dir() - # Truco: SimpleHTTPRequestHandler sirve el directorio actual, - # asegurarse de que el CWD es correcto o mover el archivo a una carpeta 'public' - # Para este ejemplo simple, asumimos que se ejecuta donde se genera el archivo. + # Handler que sirve solo el directorio 'public' + handler = partial(SimpleHTTPRequestHandler, directory=str(public_dir)) server_address = ("", settings.port) - httpd = ThreadingHTTPServer(server_address, SimpleHTTPRequestHandler) + httpd = ThreadingHTTPServer(server_address, handler) logger.info( f"Servidor M3U activo en http://localhost:{settings.port}/{settings.output_file}" diff --git a/tests/test_server.py b/tests/test_server.py index 87650d1..ab2fdda 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,5 +1,7 @@ """Tests unitarios para el módulo server.""" +import tempfile +from pathlib import Path from unittest.mock import MagicMock, patch @@ -15,165 +17,228 @@ class TestRunServer: 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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 9999 + mock_settings.output_file = "test.m3u" - from m3u_list_builder.server import run_server + 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 - run_server() + from m3u_list_builder.server import 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) + 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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - from m3u_list_builder.server import run_server + 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 - run_server() + from m3u_list_builder.server import run_server - call_args = mock_server_class.call_args[0] - assert call_args[0][1] == 8080 + 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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - from m3u_list_builder.server import run_server + 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 - run_server() + from m3u_list_builder.server import run_server - mock_server.serve_forever.assert_called_once() + 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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - from m3u_list_builder.server import run_server + 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 - # No debe lanzar excepción - run_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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - from m3u_list_builder.server import run_server + 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 - run_server() + from m3u_list_builder.server import run_server - mock_server.server_close.assert_called_once() + 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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - from m3u_list_builder.server import run_server + 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 - run_server() + from m3u_list_builder.server import run_server - mock_server.server_close.assert_called_once() + 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 tempfile.TemporaryDirectory() as tmpdir: 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 + "m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir) + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - import logging + 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 - with caplog.at_level(logging.INFO): - from m3u_list_builder.server import run_server + import logging - run_server() + with caplog.at_level(logging.INFO): + from m3u_list_builder.server import 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" + run_server() + def test_run_server_serves_from_public_dir(self): + """Test: run_server sirve desde el directorio public.""" + with tempfile.TemporaryDirectory() as tmpdir: + public_path = Path(tmpdir) 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 + "m3u_list_builder.server.get_public_dir", return_value=public_path + ): + with patch("m3u_list_builder.server.settings") as mock_settings: + mock_settings.port = 8080 + mock_settings.output_file = "playlist.m3u" - from m3u_list_builder.server import run_server + with patch( + "m3u_list_builder.server.ThreadingHTTPServer" + ) as mock_server_class: + with patch( + "m3u_list_builder.server.partial" + ) as mock_partial: + mock_server = MagicMock() + mock_server.serve_forever.side_effect = KeyboardInterrupt() + mock_server_class.return_value = mock_server - run_server() + from m3u_list_builder.server import run_server - # Verificar que se pasó el handler correcto - call_args = mock_server_class.call_args[0] - assert call_args[1] == mock_handler + run_server() + + # Verificar que partial se llamó con el directorio correcto + mock_partial.assert_called_once() + call_kwargs = mock_partial.call_args[1] + assert call_kwargs["directory"] == str(public_path) + + +class TestGetPublicDir: + """Tests para la función get_public_dir.""" + + def test_get_public_dir_creates_directory(self, tmp_path, monkeypatch): + """Test: get_public_dir crea el directorio si no existe.""" + import m3u_list_builder.server as server_module + + test_public = tmp_path / "test_public" + monkeypatch.setattr(server_module, "PUBLIC_DIR", test_public) + + from m3u_list_builder.server import get_public_dir + + result = get_public_dir() + + assert result == test_public + assert test_public.exists() + + def test_get_public_dir_returns_existing_directory(self, tmp_path, monkeypatch): + """Test: get_public_dir retorna directorio existente.""" + import m3u_list_builder.server as server_module + + test_public = tmp_path / "existing_public" + test_public.mkdir() + monkeypatch.setattr(server_module, "PUBLIC_DIR", test_public) + + from m3u_list_builder.server import get_public_dir + + result = get_public_dir() + + assert result == test_public From 0ec25115e29e4625adc6a5244431010b41fcca1a Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:19:57 +0000 Subject: [PATCH 11/15] Update Dockerfile to create public directory for playlists and correct entry point command --- Dockerfile | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index ae6429c..7178f22 100644 --- a/Dockerfile +++ b/Dockerfile @@ -31,9 +31,15 @@ COPY --from=builder /app/.venv /app/.venv # Copiar el código fuente COPY ./src /app/src +# Crear directorio public para la playlist +RUN mkdir -p /app/public + # Usuario no privilegiado por seguridad RUN useradd -m appuser && chown -R appuser /app USER appuser +# Exponer el puerto por defecto +EXPOSE 8080 + # Punto de entrada -CMD ["python", "-m", "my_project.main"] +CMD ["python", "-m", "m3u_list_builder.main"] From 2cec6b73f6293c579568bac7b14b4295be64d515 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:20:05 +0000 Subject: [PATCH 12/15] Add docker-compose.yml to define m3u-builder service configuration --- docker-compose.yml | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 docker-compose.yml diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..41b5fb8 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,11 @@ +services: + m3u-builder: + build: . + env_file: + - .env + ports: + - "${PORT:-8080}:${PORT:-8080}" + volumes: + # Opcional: persistir la playlist generada + - ./public:/app/public + restart: unless-stopped From b6142899e74aa58eaab3cce1437cf449bc5c5453 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:23:44 +0000 Subject: [PATCH 13/15] Add PYTHONPATH environment variable to Dockerfile for runtime --- Dockerfile | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 7178f22..7c1ed96 100644 --- a/Dockerfile +++ b/Dockerfile @@ -23,7 +23,8 @@ FROM python:3.14-slim as runtime WORKDIR /app ENV VIRTUAL_ENV=/app/.venv \ - PATH="/app/.venv/bin:$PATH" + PATH="/app/.venv/bin:$PATH" \ + PYTHONPATH="/app/src" # Copiar el entorno virtual generado en el stage anterior COPY --from=builder /app/.venv /app/.venv From f12e9ac54799d13a6426ea2a00037440167c7d94 Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:26:57 +0000 Subject: [PATCH 14/15] Remove optional volume configuration for persisting generated playlist in docker-compose.yml --- docker-compose.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 41b5fb8..8724ec9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,7 +5,4 @@ services: - .env ports: - "${PORT:-8080}:${PORT:-8080}" - volumes: - # Opcional: persistir la playlist generada - - ./public:/app/public restart: unless-stopped From d2d8546a426a4c4424ef18dd222aa586591eb9cf Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:37:28 +0000 Subject: [PATCH 15/15] Refactor CI workflow to use dynamic registry host and image name for Docker publishing --- .gitea/workflows/ci.yaml | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 2e11516..7905563 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -28,19 +28,25 @@ jobs: - name: Run Tests with Coverage run: poetry run pytest --cov --cov-report=term-missing --cov-fail-under=80 - + publish-container: needs: test-and-lint if: startsWith(github.ref, 'refs/tags/v') runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - + - name: Prepare Docker Metadata + id: meta + run: | + echo "REGISTRY_HOST=${GITEA_SERVER_URL#*://}" >> $GITHUB_OUTPUT + echo "IMAGE_NAME=$(echo ${{ gitea.repository }} | tr '[:upper:]' '[:lower:]')" >> $GITHUB_OUTPUT + env: + GITEA_SERVER_URL: ${{ gitea.server_url }} + - name: Login to Gitea Container Registry uses: docker/login-action@v2 with: - registry: ${{ gitea.server_url }} - # Nota: Quita 'https://' si server_url lo incluye y falla, suele ser solo dominio:puerto + registry: ${{ steps.meta.outputs.REGISTRY_HOST }} username: ${{ gitea.actor }} password: ${{ secrets.GITHUB_TOKEN }} @@ -50,5 +56,5 @@ jobs: context: . push: true tags: | - ${{ gitea.server_url }}/${{ gitea.repository }}:latest - ${{ gitea.server_url }}/${{ gitea.repository }}:${{ gitea.ref_name }} + ${{ steps.meta.outputs.REGISTRY_HOST }}/${{ steps.meta.outputs.IMAGE_NAME }}:latest + ${{ steps.meta.outputs.REGISTRY_HOST }}/${{ steps.meta.outputs.IMAGE_NAME }}:${{ gitea.ref_name }}