generated from unai/python_boilerplate
Dev #2
@ -1,22 +1,31 @@
|
||||
import logging
|
||||
from functools import partial
|
||||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
|
||||
from m3u_list_builder.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Directorio dedicado para servir archivos (solo playlist)
|
||||
PUBLIC_DIR = Path("public")
|
||||
|
||||
|
||||
def get_public_dir() -> Path:
|
||||
"""Retorna el directorio público, creándolo si no existe."""
|
||||
PUBLIC_DIR.mkdir(exist_ok=True)
|
||||
return PUBLIC_DIR
|
||||
|
||||
|
||||
def run_server():
|
||||
"""Inicia el servidor HTTP bloqueante."""
|
||||
_ = partial(SimpleHTTPRequestHandler, directory=".")
|
||||
public_dir = get_public_dir()
|
||||
|
||||
# Truco: SimpleHTTPRequestHandler sirve el directorio actual,
|
||||
# asegurarse de que el CWD es correcto o mover el archivo a una carpeta 'public'
|
||||
# Para este ejemplo simple, asumimos que se ejecuta donde se genera el archivo.
|
||||
# Handler que sirve solo el directorio 'public'
|
||||
handler = partial(SimpleHTTPRequestHandler, directory=str(public_dir))
|
||||
|
||||
server_address = ("", settings.port)
|
||||
httpd = ThreadingHTTPServer(server_address, SimpleHTTPRequestHandler)
|
||||
httpd = ThreadingHTTPServer(server_address, handler)
|
||||
|
||||
logger.info(
|
||||
f"Servidor M3U activo en http://localhost:{settings.port}/{settings.output_file}"
|
||||
|
||||
@ -1,5 +1,7 @@
|
||||
"""Tests unitarios para el módulo server."""
|
||||
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
@ -15,165 +17,228 @@ class TestRunServer:
|
||||
|
||||
def test_run_server_creates_server_with_correct_address(self):
|
||||
"""Test: run_server crea servidor en la dirección correcta."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 9999
|
||||
mock_settings.output_file = "test.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 9999
|
||||
mock_settings.output_file = "test.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
# Verificar que se creó con la dirección correcta
|
||||
mock_server_class.assert_called_once()
|
||||
call_args = mock_server_class.call_args[0]
|
||||
assert call_args[0] == ("", 9999)
|
||||
run_server()
|
||||
|
||||
# Verificar que se creó con la dirección correcta
|
||||
mock_server_class.assert_called_once()
|
||||
call_args = mock_server_class.call_args[0]
|
||||
assert call_args[0] == ("", 9999)
|
||||
|
||||
def test_run_server_uses_correct_port(self):
|
||||
"""Test: run_server usa el puerto de settings."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
call_args = mock_server_class.call_args[0]
|
||||
assert call_args[0][1] == 8080
|
||||
run_server()
|
||||
|
||||
call_args = mock_server_class.call_args[0]
|
||||
assert call_args[0][1] == 8080
|
||||
|
||||
def test_run_server_calls_serve_forever(self):
|
||||
"""Test: run_server llama a serve_forever."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
mock_server.serve_forever.assert_called_once()
|
||||
run_server()
|
||||
|
||||
mock_server.serve_forever.assert_called_once()
|
||||
|
||||
def test_run_server_handles_keyboard_interrupt(self):
|
||||
"""Test: run_server maneja KeyboardInterrupt sin crash."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
# No debe lanzar excepción
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
# No debe lanzar excepción
|
||||
run_server()
|
||||
|
||||
def test_run_server_closes_server_on_interrupt(self):
|
||||
"""Test: run_server cierra el servidor tras KeyboardInterrupt."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
mock_server.server_close.assert_called_once()
|
||||
run_server()
|
||||
|
||||
mock_server.server_close.assert_called_once()
|
||||
|
||||
def test_run_server_closes_server_on_normal_exit(self):
|
||||
"""Test: run_server cierra el servidor en salida normal."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
# Simular que serve_forever termina normalmente
|
||||
mock_server.serve_forever.return_value = None
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
# Simular que serve_forever termina normalmente
|
||||
mock_server.serve_forever.return_value = None
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
mock_server.server_close.assert_called_once()
|
||||
run_server()
|
||||
|
||||
mock_server.server_close.assert_called_once()
|
||||
|
||||
def test_run_server_logs_startup_message(self, caplog):
|
||||
"""Test: run_server registra mensaje de inicio."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=Path(tmpdir)
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
import logging
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
from m3u_list_builder.server import run_server
|
||||
import logging
|
||||
|
||||
run_server()
|
||||
with caplog.at_level(logging.INFO):
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
# El mensaje puede estar en el log
|
||||
# Nota: Dependiendo del orden de imports, puede que no se capture
|
||||
|
||||
def test_run_server_uses_simple_http_handler(self):
|
||||
"""Test: run_server usa SimpleHTTPRequestHandler."""
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
run_server()
|
||||
|
||||
def test_run_server_serves_from_public_dir(self):
|
||||
"""Test: run_server sirve desde el directorio public."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
public_path = Path(tmpdir)
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
with patch(
|
||||
"m3u_list_builder.server.SimpleHTTPRequestHandler"
|
||||
) as mock_handler:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
"m3u_list_builder.server.get_public_dir", return_value=public_path
|
||||
):
|
||||
with patch("m3u_list_builder.server.settings") as mock_settings:
|
||||
mock_settings.port = 8080
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
|
||||
from m3u_list_builder.server import run_server
|
||||
with patch(
|
||||
"m3u_list_builder.server.ThreadingHTTPServer"
|
||||
) as mock_server_class:
|
||||
with patch(
|
||||
"m3u_list_builder.server.partial"
|
||||
) as mock_partial:
|
||||
mock_server = MagicMock()
|
||||
mock_server.serve_forever.side_effect = KeyboardInterrupt()
|
||||
mock_server_class.return_value = mock_server
|
||||
|
||||
run_server()
|
||||
from m3u_list_builder.server import run_server
|
||||
|
||||
# Verificar que se pasó el handler correcto
|
||||
call_args = mock_server_class.call_args[0]
|
||||
assert call_args[1] == mock_handler
|
||||
run_server()
|
||||
|
||||
# Verificar que partial se llamó con el directorio correcto
|
||||
mock_partial.assert_called_once()
|
||||
call_kwargs = mock_partial.call_args[1]
|
||||
assert call_kwargs["directory"] == str(public_path)
|
||||
|
||||
|
||||
class TestGetPublicDir:
|
||||
"""Tests para la función get_public_dir."""
|
||||
|
||||
def test_get_public_dir_creates_directory(self, tmp_path, monkeypatch):
|
||||
"""Test: get_public_dir crea el directorio si no existe."""
|
||||
import m3u_list_builder.server as server_module
|
||||
|
||||
test_public = tmp_path / "test_public"
|
||||
monkeypatch.setattr(server_module, "PUBLIC_DIR", test_public)
|
||||
|
||||
from m3u_list_builder.server import get_public_dir
|
||||
|
||||
result = get_public_dir()
|
||||
|
||||
assert result == test_public
|
||||
assert test_public.exists()
|
||||
|
||||
def test_get_public_dir_returns_existing_directory(self, tmp_path, monkeypatch):
|
||||
"""Test: get_public_dir retorna directorio existente."""
|
||||
import m3u_list_builder.server as server_module
|
||||
|
||||
test_public = tmp_path / "existing_public"
|
||||
test_public.mkdir()
|
||||
monkeypatch.setattr(server_module, "PUBLIC_DIR", test_public)
|
||||
|
||||
from m3u_list_builder.server import get_public_dir
|
||||
|
||||
result = get_public_dir()
|
||||
|
||||
assert result == test_public
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user