generated from unai/python_boilerplate
Compare commits
No commits in common. "8124949d2edd3c24f6107405574d3c8fc020e8b2" and "659c6350d5609e29aa8656941521996d6182ec60" have entirely different histories.
8124949d2e
...
659c6350d5
@ -2,7 +2,7 @@ name: CI/CD Pipeline
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: ['**']
|
||||
branches: [main]
|
||||
tags: ['v*']
|
||||
pull_request:
|
||||
|
||||
@ -27,26 +27,20 @@ jobs:
|
||||
run: poetry run ruff check .
|
||||
|
||||
- name: Run Tests with Coverage
|
||||
run: poetry run pytest --cov --cov-report=term-missing --cov-fail-under=80
|
||||
|
||||
run: poetry run pytest --cov=my_project --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: ${{ steps.meta.outputs.REGISTRY_HOST }}
|
||||
registry: ${{ gitea.server_url }}
|
||||
# Nota: Quita 'https://' si server_url lo incluye y falla, suele ser solo dominio:puerto
|
||||
username: ${{ gitea.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
@ -56,5 +50,5 @@ jobs:
|
||||
context: .
|
||||
push: true
|
||||
tags: |
|
||||
${{ steps.meta.outputs.REGISTRY_HOST }}/${{ steps.meta.outputs.IMAGE_NAME }}:latest
|
||||
${{ steps.meta.outputs.REGISTRY_HOST }}/${{ steps.meta.outputs.IMAGE_NAME }}:${{ gitea.ref_name }}
|
||||
${{ gitea.server_url }}/${{ gitea.repository }}:latest
|
||||
${{ gitea.server_url }}/${{ gitea.repository }}:${{ gitea.ref_name }}
|
||||
|
||||
3
.gitignore
vendored
3
.gitignore
vendored
@ -42,6 +42,3 @@ Thumbs.db
|
||||
|
||||
# --- Logs ---
|
||||
*.log
|
||||
|
||||
# --- Archivos Generados ---
|
||||
public/
|
||||
|
||||
11
Dockerfile
11
Dockerfile
@ -23,8 +23,7 @@ FROM python:3.14-slim as runtime
|
||||
|
||||
WORKDIR /app
|
||||
ENV VIRTUAL_ENV=/app/.venv \
|
||||
PATH="/app/.venv/bin:$PATH" \
|
||||
PYTHONPATH="/app/src"
|
||||
PATH="/app/.venv/bin:$PATH"
|
||||
|
||||
# Copiar el entorno virtual generado en el stage anterior
|
||||
COPY --from=builder /app/.venv /app/.venv
|
||||
@ -32,15 +31,9 @@ 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", "m3u_list_builder.main"]
|
||||
CMD ["python", "-m", "my_project.main"]
|
||||
|
||||
@ -1,8 +0,0 @@
|
||||
services:
|
||||
m3u-builder:
|
||||
build: .
|
||||
env_file:
|
||||
- .env
|
||||
ports:
|
||||
- "${PORT:-8080}:${PORT:-8080}"
|
||||
restart: unless-stopped
|
||||
@ -1,10 +1,10 @@
|
||||
[tool.poetry]
|
||||
name = "m3u_list_builder"
|
||||
name = "my-project"
|
||||
version = "0.0.0"
|
||||
description = "Python tool to build M3U lists from various sources."
|
||||
description = "Python boilerplate for Gitea with Docker and Poetry"
|
||||
authors = ["Unai Blazquez <unaibg2000@gmail.com>"]
|
||||
readme = "README.md"
|
||||
packages = [{ include = "m3u_list_builder", from = "src" }]
|
||||
packages = [{include = "my_project", from = "src"}]
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python = "^3.14"
|
||||
@ -24,18 +24,14 @@ 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]
|
||||
@ -43,5 +39,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
|
||||
|
||||
@ -1,12 +1,13 @@
|
||||
"""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 m3u_list_builder.config import settings
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
from m3u_list_builder.server import run_server
|
||||
from my_project.config import settings
|
||||
from my_project.playlist import PlaylistManager
|
||||
from my_project.server import run_server
|
||||
|
||||
# Configuración básica de logging
|
||||
logging.basicConfig(
|
||||
|
||||
@ -5,23 +5,16 @@ import time
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
|
||||
from m3u_list_builder.config import settings
|
||||
from my_project.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."""
|
||||
@ -51,8 +44,9 @@ class PlaylistManager:
|
||||
|
||||
def _write_m3u(self, channels: list):
|
||||
"""Escribe el archivo M3U en disco de forma atómica."""
|
||||
temp_file = PUBLIC_DIR / f"{settings.output_file}.tmp"
|
||||
final_file = PUBLIC_DIR / settings.output_file
|
||||
# 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)
|
||||
|
||||
with open(temp_file, "w", encoding="utf-8") as f:
|
||||
f.write("#EXTM3U\n")
|
||||
@ -63,16 +57,11 @@ class PlaylistManager:
|
||||
cat_id = channel.get("category_id", "")
|
||||
|
||||
# Construir URL directa
|
||||
stream_url = (
|
||||
f"{settings.host}/live/{settings.username}/"
|
||||
f"{settings.password}/{stream_id}.ts"
|
||||
)
|
||||
stream_url = f"{settings.host}/live/{settings.username}/{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(
|
||||
f'#EXTINF:-1 tvg-id="{name}" tvg-logo="{icon}" group-title="Cat_{cat_id}",{name}\n'
|
||||
)
|
||||
f.write(extinf_line)
|
||||
f.write(f"{stream_url}\n")
|
||||
|
||||
# Reemplazo atómico
|
||||
|
||||
@ -1,31 +1,22 @@
|
||||
import logging
|
||||
from functools import partial
|
||||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
|
||||
from m3u_list_builder.config import settings
|
||||
from my_project.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."""
|
||||
public_dir = get_public_dir()
|
||||
handler = partial(SimpleHTTPRequestHandler, directory=".")
|
||||
|
||||
# Handler que sirve solo el directorio 'public'
|
||||
handler = partial(SimpleHTTPRequestHandler, directory=str(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.
|
||||
|
||||
server_address = ("", settings.port)
|
||||
httpd = ThreadingHTTPServer(server_address, handler)
|
||||
httpd = ThreadingHTTPServer(server_address, SimpleHTTPRequestHandler)
|
||||
|
||||
logger.info(
|
||||
f"Servidor M3U activo en http://localhost:{settings.port}/{settings.output_file}"
|
||||
|
||||
@ -1,55 +0,0 @@
|
||||
"""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()
|
||||
@ -1,177 +0,0 @@
|
||||
"""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")
|
||||
@ -1,181 +0,0 @@
|
||||
"""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
|
||||
@ -1,461 +0,0 @@
|
||||
"""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
|
||||
@ -1,244 +0,0 @@
|
||||
"""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
|
||||
Loading…
x
Reference in New Issue
Block a user