19 Commits

Author SHA1 Message Date
7c191b7ecd Updated package access token, again
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 39s
CI/CD Pipeline / test-and-lint (pull_request) Successful in 32s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-01 22:11:14 +00:00
73778732ba Merge pull request 'Update action user for gitea publish container' (#5) from unai-patch-3 into main
Some checks failed
CI/CD Pipeline / test-and-lint (push) Successful in 32s
CI/CD Pipeline / publish-container (push) Failing after 14s
Reviewed-on: #5
2026-02-01 21:53:53 +00:00
86f8e67318 Update action user for gitea publish container
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 32s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / test-and-lint (pull_request) Successful in 33s
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-01 21:53:10 +00:00
ed3045f6a4 Merge pull request 'Updated permissions on gitea workflow' (#4) from unai-patch-2 into main
Some checks failed
CI/CD Pipeline / test-and-lint (push) Successful in 37s
CI/CD Pipeline / publish-container (push) Failing after 16s
Reviewed-on: #4
2026-02-01 21:39:08 +00:00
a89ee4f1a0 Updated permissions on gitea workflow
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 36s
CI/CD Pipeline / test-and-lint (pull_request) Successful in 32s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-01 21:38:55 +00:00
79bbeeea00 Merge pull request 'Actualizar README.md' (#3) from unai-patch-1 into main
Some checks failed
CI/CD Pipeline / test-and-lint (push) Successful in 32s
CI/CD Pipeline / publish-container (push) Failing after 3m47s
Reviewed-on: #3
2026-02-01 19:44:45 +00:00
11668ce326 Actualizar README.md
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 36s
CI/CD Pipeline / test-and-lint (pull_request) Successful in 31s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-01 19:44:25 +00:00
8124949d2e Merge pull request 'Dev' (#2) from Dev into main
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 33s
CI/CD Pipeline / publish-container (push) Has been skipped
Reviewed-on: #2
2026-02-01 19:39:48 +00:00
d2d8546a42 Refactor CI workflow to use dynamic registry host and image name for Docker publishing
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 40s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / test-and-lint (pull_request) Successful in 36s
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-01 19:37:28 +00:00
f12e9ac547 Remove optional volume configuration for persisting generated playlist in docker-compose.yml
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 40s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:26:57 +00:00
b6142899e7 Add PYTHONPATH environment variable to Dockerfile for runtime
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 36s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:23:44 +00:00
2cec6b73f6 Add docker-compose.yml to define m3u-builder service configuration
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 36s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:20:05 +00:00
0ec25115e2 Update Dockerfile to create public directory for playlists and correct entry point command 2026-02-01 19:19:57 +00:00
3c3f2354e3 Refactor server module to use dedicated public directory and enhance tests for run_server and get_public_dir functions
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 35s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 19:14:23 +00:00
cbf22422e3 Refactor PlaylistManager to use PUBLIC_DIR for file operations and update tests accordingly 2026-02-01 19:13:42 +00:00
fd20bb28f2 Add public directory to .gitignore for generated files 2026-02-01 19:12:52 +00:00
0dc6c3607b Update test coverage source path and adjust pytest command for consistency
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 35s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:52:19 +00:00
4bf59efd6f Update Ruff configuration to exclude tests from linting
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 38s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-01 18:49:02 +00:00
659c6350d5 Merge pull request 'Dev' (#1) from Dev into main
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 26s
CI/CD Pipeline / publish-container (push) Has been skipped
Reviewed-on: #1
2026-02-01 18:18:50 +00:00
10 changed files with 454 additions and 205 deletions

View File

@@ -27,22 +27,31 @@ jobs:
run: poetry run ruff check .
- name: Run Tests with Coverage
run: poetry run pytest --cov=my_project --cov-report=term-missing --cov-fail-under=80
run: poetry run pytest --cov --cov-report=term-missing --cov-fail-under=80
publish-container:
needs: test-and-lint
if: startsWith(github.ref, 'refs/tags/v')
runs-on: ubuntu-latest
permissions:
packages: write
contents: read
steps:
- uses: actions/checkout@v3
- name: Prepare Docker Metadata
id: meta
run: |
echo "REGISTRY_HOST=${GITEA_SERVER_URL#*://}" >> $GITHUB_OUTPUT
echo "IMAGE_NAME=$(echo ${{ gitea.repository }} | tr '[:upper:]' '[:lower:]')" >> $GITHUB_OUTPUT
env:
GITEA_SERVER_URL: ${{ gitea.server_url }}
- name: Login to Gitea Container Registry
uses: docker/login-action@v2
with:
registry: ${{ gitea.server_url }}
# Nota: Quita 'https://' si server_url lo incluye y falla, suele ser solo dominio:puerto
username: ${{ gitea.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
registry: ${{ steps.meta.outputs.REGISTRY_HOST }}
username: ${{ github.actor }}
password: ${{ secrets.PACKAGES_TOKEN }}
- name: Build and Push Docker Image
uses: docker/build-push-action@v4
@@ -50,5 +59,5 @@ jobs:
context: .
push: true
tags: |
${{ gitea.server_url }}/${{ gitea.repository }}:latest
${{ gitea.server_url }}/${{ gitea.repository }}:${{ gitea.ref_name }}
${{ steps.meta.outputs.REGISTRY_HOST }}/${{ steps.meta.outputs.IMAGE_NAME }}:latest
${{ steps.meta.outputs.REGISTRY_HOST }}/${{ steps.meta.outputs.IMAGE_NAME }}:${{ gitea.ref_name }}

3
.gitignore vendored
View File

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

View File

@@ -23,7 +23,8 @@ FROM python:3.14-slim as runtime
WORKDIR /app
ENV VIRTUAL_ENV=/app/.venv \
PATH="/app/.venv/bin:$PATH"
PATH="/app/.venv/bin:$PATH" \
PYTHONPATH="/app/src"
# Copiar el entorno virtual generado en el stage anterior
COPY --from=builder /app/.venv /app/.venv
@@ -31,9 +32,15 @@ COPY --from=builder /app/.venv /app/.venv
# Copiar el código fuente
COPY ./src /app/src
# Crear directorio public para la playlist
RUN mkdir -p /app/public
# Usuario no privilegiado por seguridad
RUN useradd -m appuser && chown -R appuser /app
USER appuser
# Exponer el puerto por defecto
EXPOSE 8080
# Punto de entrada
CMD ["python", "-m", "my_project.main"]
CMD ["python", "-m", "m3u_list_builder.main"]

141
README.md
View File

@@ -0,0 +1,141 @@
# Xtream Codes M3U List Builder
[![Python 3.14+](https://img.shields.io/badge/python-3.14+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Docker](https://img.shields.io/badge/docker-ready-blue.svg)](https://www.docker.com/)
Servicio Python que genera y actualiza automáticamente listas de reproducción M3U a partir de servidores **Xtream Codes API**. Ideal para centralizar y servir playlists IPTV de forma local.
## Características
- **Actualización automática** — Regenera la playlist en intervalos configurables
- **Servidor HTTP integrado** — Sirve el archivo M3U directamente sin dependencias externas
- **Docker ready** — Despliegue simple con Docker Compose
- **Escritura atómica** — Actualiza el archivo sin interrumpir descargas activas
- **Seguridad** — Se ejecuta con usuario no privilegiado en contenedor
- **Multi-hilo** — Servidor threaded para manejar múltiples clientes simultáneos
## Inicio Rápido
### Con Docker (Recomendado)
1. **Clona el repositorio:**
```bash
git clone https://github.com/tu-usuario/xtream_codes_m3u_list_builder.git
cd xtream_codes_m3u_list_builder
```
2. **Crea un archivo `.env`:**
```env
HOST=http://tu-servidor-iptv.com
USERNAME=tu_usuario
PASSWORD=tu_contraseña
PORT=8080
UPDATE_INTERVAL=3600
OUTPUT_FILE=playlist.m3u
```
3. **Ejecuta con Docker Compose:**
```bash
docker compose up -d
```
4. **Accede a la playlist:**
```
http://localhost:8080/playlist.m3u
```
### Sin Docker
1. **Requisitos:**
- Python 3.14+
- [Poetry](https://python-poetry.org/)
2. **Instalación:**
```bash
poetry install
```
3. **Configuración:** Crea un archivo `.env` (ver ejemplo arriba) o exporta las variables de entorno.
4. **Ejecución:**
```bash
poetry run python -m m3u_list_builder.main
```
## Configuración
| Variable | Descripción | Requerido | Default |
|----------|-------------|-----------|---------|
| `HOST` | URL del servidor Xtream Codes | ✅ | — |
| `USERNAME` | Usuario del servicio IPTV | ✅ | — |
| `PASSWORD` | Contraseña del servicio IPTV | ✅ | — |
| `PORT` | Puerto del servidor HTTP local | ❌ | `8080` |
| `UPDATE_INTERVAL` | Intervalo de actualización (segundos) | ❌ | `3600` |
| `OUTPUT_FILE` | Nombre del archivo M3U generado | ❌ | `playlist.m3u` |
## Arquitectura
```
┌─────────────────────────────────────────────────────────┐
│ M3U List Builder │
├──────────────────────┬──────────────────────────────────┤
│ Updater Thread │ HTTP Server │
│ │ │
│ ┌────────────────┐ │ ┌────────────────────────────┐ │
│ │ Fetch API │ │ │ ThreadingHTTPServer │ │
│ │ (Xtream Codes) │ │ │ │ │
│ └───────┬────────┘ │ │ GET /playlist.m3u │ │
│ │ │ │ │ │
│ ┌───────▼────────┐ │ └────────────────────────────┘ │
│ │ Generate M3U │ │ │
│ │ (Atomic Write) │──┼──────► public/playlist.m3u │
│ └───────┬────────┘ │ │
│ │ │ │
│ sleep(interval) │ │
│ ↓ │ │
│ [loop] │ │
└──────────────────────┴──────────────────────────────────┘
```
## Estructura del Proyecto
```
├── src/m3u_list_builder/
│ ├── __init__.py
│ ├── config.py # Configuración con Pydantic Settings
│ ├── main.py # Punto de entrada
│ ├── playlist.py # Lógica de generación M3U
│ └── server.py # Servidor HTTP
├── tests/ # Tests unitarios
├── public/ # Directorio servido (playlist generada)
├── Dockerfile # Multi-stage build
├── docker-compose.yml
└── pyproject.toml # Dependencias y configuración
```
## Testing
```bash
# Ejecutar tests
poetry run pytest
# Con cobertura
poetry run pytest --cov
# Verificar estilo (Ruff)
poetry run ruff check src/
```
## Desarrollo
El proyecto utiliza:
- **Pydantic Settings** — Validación de configuración
- **Requests** — Cliente HTTP
- **Ruff** — Linter y formatter
- **Pytest** — Testing framework
---
<p align="center">
<sub>Desarrollado con ❤️ para la comunidad IPTV</sub>
</p>

8
docker-compose.yml Normal file
View File

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

View File

@@ -24,6 +24,7 @@ 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

View File

@@ -10,6 +10,9 @@ 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."""
@@ -17,6 +20,8 @@ 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."""
@@ -46,8 +51,8 @@ class PlaylistManager:
def _write_m3u(self, channels: list):
"""Escribe el archivo M3U en disco de forma atómica."""
temp_file = Path(f"{settings.output_file}.tmp")
final_file = Path(settings.output_file)
temp_file = PUBLIC_DIR / f"{settings.output_file}.tmp"
final_file = PUBLIC_DIR / settings.output_file
with open(temp_file, "w", encoding="utf-8") as f:
f.write("#EXTM3U\n")

View File

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

View File

@@ -219,51 +219,49 @@ class TestPlaylistManager:
def test_write_m3u_empty_channels(self, temp_dir):
"""Test: _write_m3u genera archivo con solo header para lista vacía."""
output_file = temp_dir / "playlist.m3u"
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([])
content = output_file.read_text()
assert content == "#EXTM3U\n"
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."""
output_file = temp_dir / "playlist.m3u"
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(sample_channels)
manager = PlaylistManager()
manager._write_m3u(sample_channels)
output_file = temp_dir / "playlist.m3u"
content = output_file.read_text()
content = output_file.read_text()
# Verificar header
assert content.startswith("#EXTM3U\n")
# 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
# 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",
@@ -273,52 +271,54 @@ class TestPlaylistManager:
}
]
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)
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
from m3u_list_builder.playlist import PlaylistManager
manager = PlaylistManager()
manager._write_m3u(channel)
manager = PlaylistManager()
manager._write_m3u(channel)
content = output_file.read_text()
lines = content.strip().split("\n")
output_file = temp_dir / "playlist.m3u"
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."""
output_file = temp_dir / "playlist.m3u"
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(minimal_channel)
manager = PlaylistManager()
manager._write_m3u(minimal_channel)
output_file = temp_dir / "playlist.m3u"
content = output_file.read_text()
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
# 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,22 +328,23 @@ class TestPlaylistManager:
# Crear archivo existente
output_file.write_text("OLD CONTENT")
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)
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
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,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