Compare commits

..

No commits in common. "3c3f2354e303b01b8b55f49e5ffadf89ed10f94e" and "0dc6c3607b1c2efff4973e53975572705d4db005" have entirely different histories.

5 changed files with 194 additions and 277 deletions

3
.gitignore vendored
View File

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

View File

@ -10,9 +10,6 @@ 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."""
@ -20,8 +17,6 @@ class PlaylistManager:
def __init__(self):
"""Inicialize the PlaylistManager."""
self.running = False
# Asegurar que el directorio público existe
PUBLIC_DIR.mkdir(exist_ok=True)
def fetch_and_generate(self):
"""Descarga datos y regenera el archivo M3U."""
@ -51,8 +46,8 @@ 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
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")

View File

@ -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
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()
_ = 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}"

View File

@ -219,49 +219,51 @@ class TestPlaylistManager:
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"
output_file = temp_dir / "playlist.m3u"
from m3u_list_builder.playlist import PlaylistManager
with patch("m3u_list_builder.playlist.settings") as mock_settings:
mock_settings.host = "http://test-iptv.com"
mock_settings.username = "testuser"
mock_settings.password = "testpass"
mock_settings.output_file = str(output_file)
manager = PlaylistManager()
manager._write_m3u([])
from m3u_list_builder.playlist import PlaylistManager
output_file = temp_dir / "playlist.m3u"
content = output_file.read_text()
assert content == "#EXTM3U\n"
manager = PlaylistManager()
manager._write_m3u([])
content = output_file.read_text()
assert content == "#EXTM3U\n"
def test_write_m3u_multiple_channels(self, temp_dir, sample_channels):
"""Test: _write_m3u genera archivo correcto con múltiples canales."""
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"
output_file = temp_dir / "playlist.m3u"
from m3u_list_builder.playlist import PlaylistManager
with patch("m3u_list_builder.playlist.settings") as mock_settings:
mock_settings.host = "http://test-iptv.com"
mock_settings.username = "testuser"
mock_settings.password = "testpass"
mock_settings.output_file = str(output_file)
manager = PlaylistManager()
manager._write_m3u(sample_channels)
from m3u_list_builder.playlist import PlaylistManager
output_file = temp_dir / "playlist.m3u"
content = output_file.read_text()
manager = PlaylistManager()
manager._write_m3u(sample_channels)
# Verificar header
assert content.startswith("#EXTM3U\n")
content = output_file.read_text()
# Verificar cada canal
for channel in sample_channels:
assert channel["name"] in content
assert str(channel["stream_id"]) in content
# Verificar header
assert content.startswith("#EXTM3U\n")
# Verificar cada canal
for channel in sample_channels:
assert channel["name"] in content
assert str(channel["stream_id"]) in content
def test_write_m3u_channel_format(self, temp_dir):
"""Test: _write_m3u genera formato EXTINF correcto."""
output_file = temp_dir / "playlist.m3u"
channel = [
{
"name": "Test Channel",
@ -271,54 +273,52 @@ class TestPlaylistManager:
}
]
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"
with patch("m3u_list_builder.playlist.settings") as mock_settings:
mock_settings.host = "http://iptv.com"
mock_settings.username = "user"
mock_settings.password = "pass"
mock_settings.output_file = str(output_file)
from m3u_list_builder.playlist import PlaylistManager
from m3u_list_builder.playlist import PlaylistManager
manager = PlaylistManager()
manager._write_m3u(channel)
manager = PlaylistManager()
manager._write_m3u(channel)
output_file = temp_dir / "playlist.m3u"
content = output_file.read_text()
lines = content.strip().split("\n")
content = output_file.read_text()
lines = content.strip().split("\n")
assert len(lines) == 3 # Header + EXTINF + URL
assert len(lines) == 3 # Header + EXTINF + URL
# Verificar EXTINF
assert '#EXTINF:-1 tvg-id="Test Channel"' in lines[1]
assert 'tvg-logo="http://icon.com/test.png"' in lines[1]
assert 'group-title="Cat_5"' in lines[1]
assert lines[1].endswith(",Test Channel")
# Verificar EXTINF
assert '#EXTINF:-1 tvg-id="Test Channel"' in lines[1]
assert 'tvg-logo="http://icon.com/test.png"' in lines[1]
assert 'group-title="Cat_5"' in lines[1]
assert lines[1].endswith(",Test Channel")
# Verificar URL
assert lines[2] == "http://iptv.com/live/user/pass/123.ts"
# Verificar URL
assert lines[2] == "http://iptv.com/live/user/pass/123.ts"
def test_write_m3u_missing_fields_uses_defaults(self, temp_dir, minimal_channel):
"""Test: _write_m3u usa valores por defecto para campos faltantes."""
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"
output_file = temp_dir / "playlist.m3u"
from m3u_list_builder.playlist import PlaylistManager
with patch("m3u_list_builder.playlist.settings") as mock_settings:
mock_settings.host = "http://test-iptv.com"
mock_settings.username = "testuser"
mock_settings.password = "testpass"
mock_settings.output_file = str(output_file)
manager = PlaylistManager()
manager._write_m3u(minimal_channel)
from m3u_list_builder.playlist import PlaylistManager
output_file = temp_dir / "playlist.m3u"
content = output_file.read_text()
manager = PlaylistManager()
manager._write_m3u(minimal_channel)
# Debe usar "Unknown" como nombre por defecto
assert "Unknown" in content
# Debe tener icono vacío
assert 'tvg-logo=""' in content
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)."""
@ -328,23 +328,22 @@ class TestPlaylistManager:
# 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"
with patch("m3u_list_builder.playlist.settings") as mock_settings:
mock_settings.host = "http://test-iptv.com"
mock_settings.username = "testuser"
mock_settings.password = "testpass"
mock_settings.output_file = str(output_file)
from m3u_list_builder.playlist import PlaylistManager
from m3u_list_builder.playlist import PlaylistManager
manager = PlaylistManager()
manager._write_m3u([])
manager = PlaylistManager()
manager._write_m3u([])
# El archivo temporal no debe existir después
assert not temp_file.exists()
# El archivo temporal no debe existir después
assert not temp_file.exists()
# El archivo final debe tener el nuevo contenido
assert output_file.read_text() == "#EXTM3U\n"
# El archivo final debe tener el nuevo contenido
assert output_file.read_text() == "#EXTM3U\n"
# ========================================
# Tests para loop - Complejidad Media

View File

@ -1,7 +1,5 @@
"""Tests unitarios para el módulo server."""
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
@ -17,228 +15,165 @@ class TestRunServer:
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.settings") as mock_settings:
mock_settings.port = 9999
mock_settings.output_file = "test.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
with 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
from m3u_list_builder.server import run_server
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)
# 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.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
with 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
from m3u_list_builder.server import run_server
run_server()
run_server()
call_args = mock_server_class.call_args[0]
assert call_args[0][1] == 8080
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.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
with 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
from m3u_list_builder.server import run_server
run_server()
run_server()
mock_server.serve_forever.assert_called_once()
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.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
with 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
from m3u_list_builder.server import run_server
# No debe lanzar excepción
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.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
with 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
from m3u_list_builder.server import run_server
run_server()
run_server()
mock_server.server_close.assert_called_once()
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.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"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
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
from m3u_list_builder.server import run_server
run_server()
run_server()
mock_server.server_close.assert_called_once()
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.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
with 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
import logging
with caplog.at_level(logging.INFO):
from m3u_list_builder.server import run_server
with caplog.at_level(logging.INFO):
from m3u_list_builder.server import run_server
run_server()
run_server()
# El mensaje puede estar en el log
# Nota: Dependiendo del orden de imports, puede que no se capture
def test_run_server_uses_simple_http_handler(self):
"""Test: run_server usa SimpleHTTPRequestHandler."""
with patch("m3u_list_builder.server.settings") as mock_settings:
mock_settings.port = 8080
mock_settings.output_file = "playlist.m3u"
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"
"m3u_list_builder.server.ThreadingHTTPServer"
) as mock_server_class:
with patch(
"m3u_list_builder.server.SimpleHTTPRequestHandler"
) as mock_handler:
mock_server = MagicMock()
mock_server.serve_forever.side_effect = KeyboardInterrupt()
mock_server_class.return_value = mock_server
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
from m3u_list_builder.server import run_server
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
# Verificar que se pasó el handler correcto
call_args = mock_server_class.call_args[0]
assert call_args[1] == mock_handler