diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 0c6e37a..7905563 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: @@ -27,20 +27,26 @@ 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 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 }} 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/ diff --git a/Dockerfile b/Dockerfile index ae6429c..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 @@ -31,9 +32,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"] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..8724ec9 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,8 @@ +services: + m3u-builder: + build: . + env_file: + - .env + ports: + - "${PORT:-8080}:${PORT:-8080}" + restart: unless-stopped diff --git a/pyproject.toml b/pyproject.toml index f87a622..9870a68 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" @@ -24,14 +24,18 @@ 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 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 +43,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 diff --git a/src/m3u_list_builder/main.py b/src/m3u_list_builder/main.py index a38460e..3e3797c 100644 --- a/src/m3u_list_builder/main.py +++ b/src/m3u_list_builder/main.py @@ -1,13 +1,12 @@ -""" -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 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 c417814..84ef5f7 100644 --- a/src/m3u_list_builder/playlist.py +++ b/src/m3u_list_builder/playlist.py @@ -5,16 +5,23 @@ 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__) +# Directorio dedicado para servir archivos +PUBLIC_DIR = Path("public") + class PlaylistManager: """Clase para gestionar la generación y actualización de listas M3U.""" 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.""" @@ -44,9 +51,8 @@ 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) + 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") @@ -57,11 +63,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..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 my_project.config import settings +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.""" - handler = 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/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..fb9d39a --- /dev/null +++ b/tests/test_playlist.py @@ -0,0 +1,461 @@ +"""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.""" + 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 + + manager = PlaylistManager() + manager._write_m3u([]) + + 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.""" + 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 + + manager = PlaylistManager() + manager._write_m3u(sample_channels) + + output_file = temp_dir / "playlist.m3u" + 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.""" + channel = [ + { + "name": "Test Channel", + "stream_id": 123, + "stream_icon": "http://icon.com/test.png", + "category_id": "5", + } + ] + + 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 + + manager = PlaylistManager() + manager._write_m3u(channel) + + output_file = temp_dir / "playlist.m3u" + 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.""" + 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 + + manager = PlaylistManager() + manager._write_m3u(minimal_channel) + + output_file = temp_dir / "playlist.m3u" + 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.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 + + 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..ab2fdda --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,244 @@ +"""Tests unitarios para el módulo server.""" + +import tempfile +from pathlib import Path +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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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 tempfile.TemporaryDirectory() as tmpdir: + with patch( + "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" + + 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() + + 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.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" + + 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 + + from m3u_list_builder.server import run_server + + 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