Compare commits

...

16 Commits

Author SHA1 Message Date
8124949d2e Merge pull request 'Dev' (#2) from Dev into main
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 33s
CI/CD Pipeline / publish-container (push) Has been skipped
Reviewed-on: #2
2026-02-01 19:39:48 +00:00
d2d8546a42 Refactor CI workflow to use dynamic registry host and image name for Docker publishing
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 40s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / test-and-lint (pull_request) Successful in 36s
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-01 19:37:28 +00:00
f12e9ac547 Remove optional volume configuration for persisting generated playlist in docker-compose.yml
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 40s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:26:57 +00:00
b6142899e7 Add PYTHONPATH environment variable to Dockerfile for runtime
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 36s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:23:44 +00:00
2cec6b73f6 Add docker-compose.yml to define m3u-builder service configuration
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 36s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:20:05 +00:00
0ec25115e2 Update Dockerfile to create public directory for playlists and correct entry point command 2026-02-01 19:19:57 +00:00
3c3f2354e3 Refactor server module to use dedicated public directory and enhance tests for run_server and get_public_dir functions
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 35s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:14:23 +00:00
cbf22422e3 Refactor PlaylistManager to use PUBLIC_DIR for file operations and update tests accordingly 2026-02-01 19:13:42 +00:00
fd20bb28f2 Add public directory to .gitignore for generated files 2026-02-01 19:12:52 +00:00
0dc6c3607b Update test coverage source path and adjust pytest command for consistency
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 35s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:52:19 +00:00
4bf59efd6f Update Ruff configuration to exclude tests from linting
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 38s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:49:02 +00:00
128432f679 Add unit tests for all modules including config, main, playlist, and server
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 32s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:47:32 +00:00
9f7d3d98c1 Refactor import statements to use the correct module path 2026-02-01 18:47:14 +00:00
5342644b1d Liniting
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 32s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:36:40 +00:00
d9dfe7b7c4 Update CI/CD pipeline to trigger on all branches
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 27s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:33:15 +00:00
7de3aa5b94 Update project name and description in pyproject.toml 2026-02-01 18:28:09 +00:00
13 changed files with 1200 additions and 35 deletions

View File

@ -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 }}

3
.gitignore vendored
View File

@ -42,3 +42,6 @@ Thumbs.db
# --- Logs ---
*.log
# --- Archivos Generados ---
public/

View File

@ -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"]

8
docker-compose.yml Normal file
View File

@ -0,0 +1,8 @@
services:
m3u-builder:
build: .
env_file:
- .env
ports:
- "${PORT:-8080}:${PORT:-8080}"
restart: unless-stopped

View File

@ -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 <unaibg2000@gmail.com>"]
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

View File

@ -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(

View File

@ -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

View File

@ -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}"

55
tests/conftest.py Normal file
View File

@ -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()

177
tests/test_config.py Normal file
View File

@ -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")

181
tests/test_main.py Normal file
View File

@ -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

461
tests/test_playlist.py Normal file
View File

@ -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

244
tests/test_server.py Normal file
View File

@ -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