From 3c3f2354e303b01b8b55f49e5ffadf89ed10f94e Mon Sep 17 00:00:00 2001 From: unai Date: Sun, 1 Feb 2026 19:14:23 +0000 Subject: [PATCH] Refactor server module to use dedicated public directory and enhance tests for run_server and get_public_dir functions --- src/m3u_list_builder/server.py | 19 ++- tests/test_server.py | 287 ++++++++++++++++++++------------- 2 files changed, 190 insertions(+), 116 deletions(-) diff --git a/src/m3u_list_builder/server.py b/src/m3u_list_builder/server.py index 38ac3b4..b969f63 100644 --- a/src/m3u_list_builder/server.py +++ b/src/m3u_list_builder/server.py @@ -1,22 +1,31 @@ import logging from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path from 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}" diff --git a/tests/test_server.py b/tests/test_server.py index 87650d1..ab2fdda 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -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