generated from unai/python_boilerplate
Compare commits
16 Commits
0dc6c3607b
...
v0.1.2
| Author | SHA1 | Date | |
|---|---|---|---|
| 73778732ba | |||
| 86f8e67318 | |||
| ed3045f6a4 | |||
| a89ee4f1a0 | |||
| 79bbeeea00 | |||
| 11668ce326 | |||
| 8124949d2e | |||
| d2d8546a42 | |||
| f12e9ac547 | |||
| b6142899e7 | |||
| 2cec6b73f6 | |||
| 0ec25115e2 | |||
| 3c3f2354e3 | |||
| cbf22422e3 | |||
| fd20bb28f2 | |||
| 659c6350d5 |
@@ -28,20 +28,29 @@ jobs:
|
||||
|
||||
- name: Run Tests with Coverage
|
||||
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 }}
|
||||
registry: ${{ steps.meta.outputs.REGISTRY_HOST }}
|
||||
username: ${{ github.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: Build and Push Docker Image
|
||||
@@ -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
3
.gitignore
vendored
@@ -42,3 +42,6 @@ Thumbs.db
|
||||
|
||||
# --- Logs ---
|
||||
*.log
|
||||
|
||||
# --- Archivos Generados ---
|
||||
public/
|
||||
|
||||
11
Dockerfile
11
Dockerfile
@@ -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
141
README.md
@@ -0,0 +1,141 @@
|
||||
# Xtream Codes M3U List Builder
|
||||
|
||||
[](https://www.python.org/downloads/)
|
||||
[](https://opensource.org/licenses/MIT)
|
||||
[](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
8
docker-compose.yml
Normal file
@@ -0,0 +1,8 @@
|
||||
services:
|
||||
m3u-builder:
|
||||
build: .
|
||||
env_file:
|
||||
- .env
|
||||
ports:
|
||||
- "${PORT:-8080}:${PORT:-8080}"
|
||||
restart: unless-stopped
|
||||
@@ -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")
|
||||
|
||||
@@ -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}"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user