generated from unai/python_boilerplate
Compare commits
13 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| df17eaaf12 | |||
| fb7eff451d | |||
| 9e8f3bcb47 | |||
| 34420e6758 | |||
| 545958028d | |||
| e62c243542 | |||
| 6c357dd977 | |||
| 7733c7e2ac | |||
| d95d6d6c58 | |||
| c1d9b6bf54 | |||
| 757ed91aae | |||
| 1cf174e896 | |||
| ab9ccd16aa |
30
.dockerignore
Normal file
30
.dockerignore
Normal file
@ -0,0 +1,30 @@
|
||||
# Control de versiones
|
||||
.git
|
||||
.gitignore
|
||||
|
||||
# Entornos virtuales y dependencias locales
|
||||
.venv
|
||||
venv
|
||||
env
|
||||
__pycache__
|
||||
*.pyc
|
||||
*.pyo
|
||||
*.pyd
|
||||
|
||||
# Testing y calidad
|
||||
.pytest_cache
|
||||
.coverage
|
||||
htmlcov
|
||||
.ruff_cache
|
||||
tests
|
||||
|
||||
# Configuración de IDE y dev
|
||||
.vscode
|
||||
.idea
|
||||
.devcontainer
|
||||
devcontainer.json
|
||||
|
||||
# Salidas de build
|
||||
dist
|
||||
build
|
||||
*.egg-info
|
||||
35
README.md
35
README.md
@ -8,9 +8,11 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
|
||||
|
||||
## Características
|
||||
- **Actualización automática** — Regenera la playlist en intervalos configurables
|
||||
- **Servidor HTTP integrado** — Sirve el archivo M3U directamente sin dependencias externas
|
||||
- **EPG integrado** — Descarga y filtra automáticamente la guía de programación (XMLTV)
|
||||
- **Filtros avanzados** — Incluye canales por prefijo o excluye por contenido del nombre
|
||||
- **Servidor HTTP integrado** — Sirve el archivo M3U y EPG directamente sin dependencias externas
|
||||
- **Docker ready** — Despliegue simple con Docker Compose
|
||||
- **Escritura atómica** — Actualiza el archivo sin interrumpir descargas activas
|
||||
- **Escritura atómica** — Actualiza archivos sin interrumpir descargas activas
|
||||
- **Seguridad** — Se ejecuta con usuario no privilegiado en contenedor
|
||||
- **Multi-hilo** — Servidor threaded para manejar múltiples clientes simultáneos
|
||||
|
||||
@ -32,6 +34,8 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
|
||||
PORT=8080
|
||||
UPDATE_INTERVAL=3600
|
||||
OUTPUT_FILE=playlist.m3u
|
||||
INCLUDE_TEXT="ES:"
|
||||
EXCLUDE_TEXT=["XXX","Adultos","24/7"]
|
||||
```
|
||||
|
||||
3. **Ejecuta con Docker Compose:**
|
||||
@ -39,9 +43,10 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
|
||||
docker compose up -d
|
||||
```
|
||||
|
||||
4. **Accede a la playlist:**
|
||||
4. **Accede a la playlist y EPG:**
|
||||
```
|
||||
http://localhost:8080/playlist.m3u
|
||||
http://localhost:8080/epg.xml
|
||||
```
|
||||
|
||||
### Sin Docker
|
||||
@ -72,6 +77,21 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
|
||||
| `PORT` | Puerto del servidor HTTP local | ❌ | `8080` |
|
||||
| `UPDATE_INTERVAL` | Intervalo de actualización (segundos) | ❌ | `3600` |
|
||||
| `OUTPUT_FILE` | Nombre del archivo M3U generado | ❌ | `playlist.m3u` |
|
||||
| `INCLUDE_TEXT` | Lista JSON de prefijos. Solo canales que **empiecen** con alguno | ❌ | `[]` |
|
||||
| `EXCLUDE_TEXT` | Lista JSON de textos. Excluye canales que **contengan** alguno | ❌ | `[]` |
|
||||
|
||||
### Filtros
|
||||
|
||||
- **`INCLUDE_TEXT`**: Filtra canales que **empiecen** con alguno de los textos especificados (lógica OR). Si está vacío, incluye todos los canales.
|
||||
- **`EXCLUDE_TEXT`**: Excluye canales cuyo nombre **contenga** alguno de los textos (lógica OR). Se aplica después del filtro de inclusión.
|
||||
- Los mismos filtros se aplican automáticamente al EPG.
|
||||
|
||||
Ejemplo:
|
||||
```env
|
||||
# Solo canales que empiecen con "ES:" o "UK:", excluyendo los que contengan "Adult"
|
||||
INCLUDE_TEXT=["ES:", "UK:"]
|
||||
EXCLUDE_TEXT=["Adult", "XXX"]
|
||||
```
|
||||
|
||||
## Arquitectura
|
||||
|
||||
@ -85,10 +105,15 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
|
||||
│ │ Fetch API │ │ │ ThreadingHTTPServer │ │
|
||||
│ │ (Xtream Codes) │ │ │ │ │
|
||||
│ └───────┬────────┘ │ │ GET /playlist.m3u │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ │ GET /epg.xml │ │
|
||||
│ ┌───────▼────────┐ │ └────────────────────────────┘ │
|
||||
│ │ Generate M3U │ │ │
|
||||
│ │ (Atomic Write) │──┼──────► public/playlist.m3u │
|
||||
│ │ + Apply Filter │──┼──────► public/playlist.m3u │
|
||||
│ └───────┬────────┘ │ │
|
||||
│ │ │ │
|
||||
│ ┌───────▼────────┐ │ │
|
||||
│ │ Fetch & Filter │ │ │
|
||||
│ │ EPG (XMLTV) │──┼──────► public/epg.xml │
|
||||
│ └───────┬────────┘ │ │
|
||||
│ │ │ │
|
||||
│ sleep(interval) │ │
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[tool.poetry]
|
||||
name = "m3u_list_builder"
|
||||
version = "0.0.0"
|
||||
version = "0.3.2"
|
||||
description = "Python tool to build M3U lists from various sources."
|
||||
authors = ["Unai Blazquez <unaibg2000@gmail.com>"]
|
||||
readme = "README.md"
|
||||
|
||||
@ -24,6 +24,20 @@ class Settings(BaseSettings):
|
||||
env_file=".env", env_file_encoding="utf-8", extra="ignore"
|
||||
)
|
||||
|
||||
# Filtros por prefijo de nombre de canal
|
||||
include_text: list[str] = Field(
|
||||
default_factory=list,
|
||||
description="Lista de textos obligatorios. Debe contener al menos uno.",
|
||||
)
|
||||
exclude_text: list[str] = Field(
|
||||
default_factory=list,
|
||||
description="Lista de textos prohibidos",
|
||||
)
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=".env", env_file_encoding="utf-8", extra="ignore"
|
||||
)
|
||||
|
||||
|
||||
# Instancia única de configuración
|
||||
settings = Settings()
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
|
||||
import logging
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
@ -24,7 +25,7 @@ class PlaylistManager:
|
||||
PUBLIC_DIR.mkdir(exist_ok=True)
|
||||
|
||||
def fetch_and_generate(self):
|
||||
"""Descarga datos y regenera el archivo M3U."""
|
||||
"""Descarga datos y regenera el archivo M3U y EPG."""
|
||||
logger.info("Iniciando actualización de playlist...")
|
||||
|
||||
url = f"{settings.host}/player_api.php"
|
||||
@ -44,11 +45,95 @@ class PlaylistManager:
|
||||
f"Playlist actualizada exitosamente. Total canales: {len(data)}"
|
||||
)
|
||||
|
||||
# Fetch and write EPG data
|
||||
self.fetch_and_write_epg()
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Error de red al obtener playlist: {e}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error inesperado actualizando playlist: {e}")
|
||||
|
||||
def fetch_and_write_epg(self):
|
||||
"""Descarga y guarda los datos EPG desde el servidor XMLTV."""
|
||||
logger.info("Iniciando descarga de EPG...")
|
||||
|
||||
url = f"{settings.host}/xmltv.php"
|
||||
params = {
|
||||
"username": settings.username,
|
||||
"password": settings.password,
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.get(url, params=params, timeout=60)
|
||||
response.raise_for_status()
|
||||
|
||||
# Parse and filter EPG XML
|
||||
filtered_epg = self._filter_epg(response.content)
|
||||
self._write_epg(filtered_epg)
|
||||
|
||||
logger.info("EPG actualizado exitosamente.")
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Error de red al obtener EPG: {e}")
|
||||
except ET.ParseError as e:
|
||||
logger.error(f"Error parseando XML de EPG: {e}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error inesperado actualizando EPG: {e}")
|
||||
|
||||
def _filter_epg(self, xml_content: bytes) -> ET.Element:
|
||||
"""Filtra el EPG para incluir solo canales que pasan los filtros."""
|
||||
root = ET.fromstring(xml_content)
|
||||
|
||||
# Get list of channel IDs to keep based on filters
|
||||
channels_to_remove = []
|
||||
for channel in root.findall("channel"):
|
||||
display_name = channel.find("display-name")
|
||||
name = display_name.text if display_name is not None else ""
|
||||
|
||||
should_remove = False
|
||||
|
||||
# Apply include filter (startswith)
|
||||
if settings.include_text:
|
||||
if not any(
|
||||
name.lower().startswith(inc_text.lower())
|
||||
for inc_text in settings.include_text
|
||||
):
|
||||
should_remove = True
|
||||
|
||||
# Apply exclude filter (contains)
|
||||
if not should_remove and settings.exclude_text:
|
||||
if any(
|
||||
exc_text.lower() in name.lower()
|
||||
for exc_text in settings.exclude_text
|
||||
):
|
||||
should_remove = True
|
||||
|
||||
if should_remove:
|
||||
channels_to_remove.append(channel.get("id"))
|
||||
|
||||
# Remove filtered channels
|
||||
for channel in root.findall("channel"):
|
||||
if channel.get("id") in channels_to_remove:
|
||||
root.remove(channel)
|
||||
|
||||
# Remove programmes for filtered channels
|
||||
for programme in root.findall("programme"):
|
||||
if programme.get("channel") in channels_to_remove:
|
||||
root.remove(programme)
|
||||
|
||||
return root
|
||||
|
||||
def _write_epg(self, epg_root: ET.Element):
|
||||
"""Escribe el archivo EPG en disco de forma atómica."""
|
||||
temp_file = PUBLIC_DIR / "epg.xml.tmp"
|
||||
final_file = PUBLIC_DIR / "epg.xml"
|
||||
|
||||
tree = ET.ElementTree(epg_root)
|
||||
tree.write(temp_file, encoding="utf-8", xml_declaration=True)
|
||||
|
||||
# Reemplazo atómico
|
||||
temp_file.replace(final_file)
|
||||
|
||||
def _write_m3u(self, channels: list):
|
||||
"""Escribe el archivo M3U en disco de forma atómica."""
|
||||
temp_file = PUBLIC_DIR / f"{settings.output_file}.tmp"
|
||||
@ -61,7 +146,22 @@ class PlaylistManager:
|
||||
stream_id = channel.get("stream_id")
|
||||
icon = channel.get("stream_icon", "")
|
||||
cat_id = channel.get("category_id", "")
|
||||
|
||||
epg_id = channel.get("epg_channel_id", "")
|
||||
if not epg_id:
|
||||
epg_id = name
|
||||
# Filtros de inclusión/exclusión por prefijo de nombre
|
||||
if settings.include_text:
|
||||
if not any(
|
||||
name.lower().startswith(inc_text.lower())
|
||||
for inc_text in settings.include_text
|
||||
):
|
||||
continue
|
||||
if settings.exclude_text:
|
||||
if any(
|
||||
exc_text.lower() in name.lower()
|
||||
for exc_text in settings.exclude_text
|
||||
):
|
||||
continue
|
||||
# Construir URL directa
|
||||
stream_url = (
|
||||
f"{settings.host}/live/{settings.username}/"
|
||||
@ -69,7 +169,7 @@ class PlaylistManager:
|
||||
)
|
||||
|
||||
extinf_line = (
|
||||
f'#EXTINF:-1 tvg-id="{name}" tvg-logo="{icon}" '
|
||||
f'#EXTINF:-1 tvg-id="{epg_id}" tvg-logo="{icon}" '
|
||||
f'group-title="Cat_{cat_id}",{name}\n'
|
||||
)
|
||||
f.write(extinf_line)
|
||||
|
||||
@ -28,6 +28,55 @@ def sample_channels():
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def channels_for_filtering():
|
||||
"""Fixture con canales para probar filtros de inclusión/exclusión."""
|
||||
return [
|
||||
{
|
||||
"name": "ESPN Sports",
|
||||
"stream_id": 201,
|
||||
"stream_icon": "http://example.com/espn.png",
|
||||
"category_id": "sports",
|
||||
},
|
||||
{
|
||||
"name": "HBO Movies",
|
||||
"stream_id": 202,
|
||||
"stream_icon": "http://example.com/hbo.png",
|
||||
"category_id": "movies",
|
||||
},
|
||||
{
|
||||
"name": "CNN News",
|
||||
"stream_id": 203,
|
||||
"stream_icon": "http://example.com/cnn.png",
|
||||
"category_id": "news",
|
||||
},
|
||||
{
|
||||
"name": "Sports Center ESPN",
|
||||
"stream_id": 204,
|
||||
"stream_icon": "http://example.com/sc.png",
|
||||
"category_id": "sports",
|
||||
},
|
||||
{
|
||||
"name": "BBC World",
|
||||
"stream_id": 205,
|
||||
"stream_icon": "http://example.com/bbc.png",
|
||||
"category_id": "news",
|
||||
},
|
||||
{
|
||||
"name": "Adult Content",
|
||||
"stream_id": 206,
|
||||
"stream_icon": "http://example.com/adult.png",
|
||||
"category_id": "adult",
|
||||
},
|
||||
{
|
||||
"name": "FOX News",
|
||||
"stream_id": 207,
|
||||
"stream_icon": "http://example.com/fox.png",
|
||||
"category_id": "news",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def empty_channels():
|
||||
"""Fixture con lista vacía de canales."""
|
||||
@ -51,5 +100,7 @@ def mock_settings(monkeypatch):
|
||||
port = 8080
|
||||
update_interval = 60
|
||||
output_file = "test_playlist.m3u"
|
||||
include_text = []
|
||||
exclude_text = []
|
||||
|
||||
return MockSettings()
|
||||
|
||||
@ -175,3 +175,133 @@ class TestSettings:
|
||||
# No debe lanzar excepción
|
||||
settings = Settings()
|
||||
assert not hasattr(settings, "unknown_field")
|
||||
|
||||
# ========================================
|
||||
# Tests para include_text y exclude_text
|
||||
# ========================================
|
||||
|
||||
def test_settings_include_text_default_empty_list(self, monkeypatch):
|
||||
"""Test: include_text tiene valor por defecto lista vacía."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.include_text == []
|
||||
assert isinstance(settings.include_text, list)
|
||||
|
||||
def test_settings_exclude_text_default_empty_list(self, monkeypatch):
|
||||
"""Test: exclude_text tiene valor por defecto lista vacía."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.exclude_text == []
|
||||
assert isinstance(settings.exclude_text, list)
|
||||
|
||||
def test_settings_include_text_from_env_json(self, monkeypatch):
|
||||
"""Test: include_text acepta formato JSON desde variable de entorno."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
monkeypatch.setenv("INCLUDE_TEXT", '["ESPN", "CNN"]')
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.include_text == ["ESPN", "CNN"]
|
||||
|
||||
def test_settings_exclude_text_from_env_json(self, monkeypatch):
|
||||
"""Test: exclude_text acepta formato JSON desde variable de entorno."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
monkeypatch.setenv("EXCLUDE_TEXT", '["Adult", "XXX"]')
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.exclude_text == ["Adult", "XXX"]
|
||||
|
||||
def test_settings_include_and_exclude_together(self, monkeypatch):
|
||||
"""Test: include_text y exclude_text pueden usarse juntos."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
monkeypatch.setenv("INCLUDE_TEXT", '["Sports", "News"]')
|
||||
monkeypatch.setenv("EXCLUDE_TEXT", '["Adult"]')
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.include_text == ["Sports", "News"]
|
||||
assert settings.exclude_text == ["Adult"]
|
||||
|
||||
def test_settings_include_text_single_value(self, monkeypatch):
|
||||
"""Test: include_text acepta un solo valor en lista."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
monkeypatch.setenv("INCLUDE_TEXT", '["OnlyThis"]')
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.include_text == ["OnlyThis"]
|
||||
assert len(settings.include_text) == 1
|
||||
|
||||
def test_settings_exclude_text_single_value(self, monkeypatch):
|
||||
"""Test: exclude_text acepta un solo valor en lista."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
monkeypatch.setenv("EXCLUDE_TEXT", '["BlockThis"]')
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.exclude_text == ["BlockThis"]
|
||||
assert len(settings.exclude_text) == 1
|
||||
|
||||
def test_settings_include_text_empty_json_array(self, monkeypatch):
|
||||
"""Test: include_text acepta array JSON vacío."""
|
||||
monkeypatch.setenv("HOST", "http://test.com")
|
||||
monkeypatch.setenv("USERNAME", "user")
|
||||
monkeypatch.setenv("PASSWORD", "pass")
|
||||
monkeypatch.setenv("INCLUDE_TEXT", "[]")
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=None)
|
||||
|
||||
assert settings.include_text == []
|
||||
|
||||
def test_settings_from_env_file_with_filters(self, tmp_path, monkeypatch):
|
||||
"""Test: Settings lee filtros desde archivo .env."""
|
||||
env_file = tmp_path / ".env"
|
||||
env_file.write_text(
|
||||
'HOST=http://test.com\n'
|
||||
'USERNAME=user\n'
|
||||
'PASSWORD=pass\n'
|
||||
'INCLUDE_TEXT=["HBO", "ESPN"]\n'
|
||||
'EXCLUDE_TEXT=["Adult"]\n'
|
||||
)
|
||||
|
||||
from m3u_list_builder.config import Settings
|
||||
|
||||
settings = Settings(_env_file=str(env_file))
|
||||
|
||||
assert settings.include_text == ["HBO", "ESPN"]
|
||||
assert settings.exclude_text == ["Adult"]
|
||||
|
||||
@ -66,10 +66,11 @@ class TestPlaylistManager:
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with patch.object(manager, "_write_m3u") as mock_write:
|
||||
manager.fetch_and_generate()
|
||||
with patch.object(manager, "fetch_and_write_epg"):
|
||||
manager.fetch_and_generate()
|
||||
|
||||
mock_get.assert_called_once()
|
||||
mock_write.assert_called_once_with(sample_channels)
|
||||
mock_get.assert_called_once()
|
||||
mock_write.assert_called_once_with(sample_channels)
|
||||
|
||||
def test_fetch_and_generate_correct_api_call(self):
|
||||
"""Test: fetch_and_generate llama a la API correctamente."""
|
||||
@ -89,17 +90,18 @@ class TestPlaylistManager:
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with patch.object(manager, "_write_m3u"):
|
||||
manager.fetch_and_generate()
|
||||
with patch.object(manager, "fetch_and_write_epg"):
|
||||
manager.fetch_and_generate()
|
||||
|
||||
mock_get.assert_called_once_with(
|
||||
"http://test-iptv.com/player_api.php",
|
||||
params={
|
||||
"username": "testuser",
|
||||
"password": "testpass",
|
||||
"action": "get_live_streams",
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
mock_get.assert_called_once_with(
|
||||
"http://test-iptv.com/player_api.php",
|
||||
params={
|
||||
"username": "testuser",
|
||||
"password": "testpass",
|
||||
"action": "get_live_streams",
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
def test_fetch_and_generate_request_exception(self, caplog):
|
||||
"""Test: fetch_and_generate maneja RequestException sin crash."""
|
||||
@ -243,6 +245,8 @@ class TestPlaylistManager:
|
||||
mock_settings.username = "testuser"
|
||||
mock_settings.password = "testpass"
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
@ -277,6 +281,8 @@ class TestPlaylistManager:
|
||||
mock_settings.username = "user"
|
||||
mock_settings.password = "pass"
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
@ -298,6 +304,104 @@ class TestPlaylistManager:
|
||||
# Verificar URL
|
||||
assert lines[2] == "http://iptv.com/live/user/pass/123.ts"
|
||||
|
||||
def test_write_m3u_uses_epg_channel_id_when_present(self, temp_dir):
|
||||
"""Test: _write_m3u usa epg_channel_id en tvg-id cuando está presente."""
|
||||
channel = [
|
||||
{
|
||||
"name": "Test Channel",
|
||||
"stream_id": 123,
|
||||
"stream_icon": "http://icon.com/test.png",
|
||||
"category_id": "5",
|
||||
"epg_channel_id": "test.channel.epg",
|
||||
}
|
||||
]
|
||||
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channel)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Debe usar epg_channel_id en tvg-id
|
||||
assert 'tvg-id="test.channel.epg"' in content
|
||||
# El nombre debe seguir apareciendo al final de EXTINF
|
||||
assert ",Test Channel" in content
|
||||
|
||||
def test_write_m3u_falls_back_to_name_when_epg_channel_id_empty(self, temp_dir):
|
||||
"""Test: _write_m3u usa name como tvg-id cuando epg_channel_id está vacío."""
|
||||
channel = [
|
||||
{
|
||||
"name": "Fallback Channel",
|
||||
"stream_id": 456,
|
||||
"stream_icon": "",
|
||||
"category_id": "1",
|
||||
"epg_channel_id": "",
|
||||
}
|
||||
]
|
||||
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channel)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Debe usar name como fallback en tvg-id
|
||||
assert 'tvg-id="Fallback Channel"' in content
|
||||
|
||||
def test_write_m3u_falls_back_to_name_when_epg_channel_id_missing(self, temp_dir):
|
||||
"""Test: _write_m3u usa name como tvg-id cuando epg_channel_id no existe."""
|
||||
channel = [
|
||||
{
|
||||
"name": "No EPG Channel",
|
||||
"stream_id": 789,
|
||||
"stream_icon": "",
|
||||
"category_id": "2",
|
||||
# No epg_channel_id field at all
|
||||
}
|
||||
]
|
||||
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channel)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Debe usar name como fallback en tvg-id
|
||||
assert 'tvg-id="No EPG Channel"' in content
|
||||
|
||||
def test_write_m3u_missing_fields_uses_defaults(self, temp_dir, minimal_channel):
|
||||
"""Test: _write_m3u usa valores por defecto para campos faltantes."""
|
||||
with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
|
||||
@ -306,6 +410,8 @@ class TestPlaylistManager:
|
||||
mock_settings.username = "testuser"
|
||||
mock_settings.password = "testpass"
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
@ -459,3 +565,784 @@ class TestPlaylistManager:
|
||||
manager.loop()
|
||||
|
||||
assert iterations == 3
|
||||
|
||||
# ========================================
|
||||
# Tests para filtros include_text y exclude_text
|
||||
# Complejidad Alta - Múltiples paths de filtrado
|
||||
# ========================================
|
||||
|
||||
def test_write_m3u_include_text_filters_by_prefix(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: include_text filtra canales que EMPIEZAN con el texto."""
|
||||
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"
|
||||
mock_settings.include_text = ["ESPN"]
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "ESPN Sports" empieza con "ESPN" - DEBE incluirse
|
||||
assert "ESPN Sports" in content
|
||||
# "Sports Center ESPN" NO empieza con "ESPN" - NO debe incluirse
|
||||
assert "Sports Center ESPN" not in content
|
||||
# Otros canales no deben estar
|
||||
assert "HBO Movies" not in content
|
||||
assert "CNN News" not in content
|
||||
|
||||
def test_write_m3u_include_text_case_insensitive(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: include_text es case-insensitive."""
|
||||
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"
|
||||
mock_settings.include_text = ["espn"] # minúsculas
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "ESPN Sports" debe incluirse aunque el filtro está en minúsculas
|
||||
assert "ESPN Sports" in content
|
||||
|
||||
def test_write_m3u_include_text_multiple_prefixes(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: include_text acepta múltiples prefijos (OR logic)."""
|
||||
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"
|
||||
mock_settings.include_text = ["ESPN", "CNN", "BBC"]
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Canales que empiezan con ESPN, CNN o BBC
|
||||
assert "ESPN Sports" in content
|
||||
assert "CNN News" in content
|
||||
assert "BBC World" in content
|
||||
# Canales que no empiezan con ninguno
|
||||
assert "HBO Movies" not in content
|
||||
assert "FOX News" not in content
|
||||
|
||||
def test_write_m3u_include_text_empty_includes_all(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: include_text vacío incluye todos los canales."""
|
||||
with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
|
||||
with patch("m3u_list_builder.playlist.settings") as mock_settings:
|
||||
mock_settings.host = "http://test-iptv.com"
|
||||
mock_settings.username = "testuser"
|
||||
mock_settings.password = "testpass"
|
||||
mock_settings.output_file = "playlist.m3u"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Todos los canales deben estar incluidos
|
||||
for channel in channels_for_filtering:
|
||||
assert channel["name"] in content
|
||||
|
||||
def test_write_m3u_exclude_text_filters_by_contains(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: exclude_text filtra canales que CONTIENEN el texto."""
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = ["Adult"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "Adult Content" contiene "Adult" - NO debe incluirse
|
||||
assert "Adult Content" not in content
|
||||
# Otros canales deben estar
|
||||
assert "ESPN Sports" in content
|
||||
assert "HBO Movies" in content
|
||||
|
||||
def test_write_m3u_exclude_text_matches_anywhere_in_name(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: exclude_text busca en cualquier parte del nombre."""
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = ["News"] # Está en medio/final
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "CNN News" y "FOX News" contienen "News" - NO deben incluirse
|
||||
assert "CNN News" not in content
|
||||
assert "FOX News" not in content
|
||||
# Otros canales deben estar
|
||||
assert "ESPN Sports" in content
|
||||
assert "HBO Movies" in content
|
||||
|
||||
def test_write_m3u_exclude_text_case_insensitive(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: exclude_text es case-insensitive."""
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = ["adult"] # minúsculas
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "Adult Content" debe excluirse aunque el filtro está en minúsculas
|
||||
assert "Adult Content" not in content
|
||||
|
||||
def test_write_m3u_exclude_text_multiple_patterns(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: exclude_text acepta múltiples patrones (OR logic)."""
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
mock_settings.exclude_text = ["Adult", "News"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Canales que contienen "Adult" o "News" no deben incluirse
|
||||
assert "Adult Content" not in content
|
||||
assert "CNN News" not in content
|
||||
assert "FOX News" not in content
|
||||
# Otros canales deben estar
|
||||
assert "ESPN Sports" in content
|
||||
assert "HBO Movies" in content
|
||||
assert "BBC World" in content
|
||||
|
||||
def test_write_m3u_include_and_exclude_combined(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: include_text y exclude_text funcionan juntos."""
|
||||
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"
|
||||
# Solo canales que empiezan con "ESPN" o "Sports"
|
||||
mock_settings.include_text = ["ESPN", "Sports"]
|
||||
# Pero excluir los que contienen "Center"
|
||||
mock_settings.exclude_text = ["Center"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "ESPN Sports" empieza con "ESPN" y no contiene "Center" - INCLUIR
|
||||
assert "ESPN Sports" in content
|
||||
# "Sports Center ESPN" empieza con "Sports" pero contiene "Center" - EXCLUIR
|
||||
assert "Sports Center ESPN" not in content
|
||||
|
||||
def test_write_m3u_include_filters_before_exclude(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: include se aplica antes que exclude."""
|
||||
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"
|
||||
mock_settings.include_text = ["CNN"]
|
||||
mock_settings.exclude_text = ["News"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# "CNN News" empieza con "CNN" pero contiene "News" - EXCLUIR
|
||||
assert "CNN News" not in content
|
||||
# Solo debe quedar el header
|
||||
assert content.strip() == "#EXTM3U"
|
||||
|
||||
def test_write_m3u_no_match_for_include_results_in_empty(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: si ningún canal coincide con include, resultado vacío."""
|
||||
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"
|
||||
mock_settings.include_text = ["NONEXISTENT"]
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Solo header, sin canales
|
||||
assert content == "#EXTM3U\n"
|
||||
|
||||
def test_write_m3u_exclude_all_results_in_empty(
|
||||
self, temp_dir, channels_for_filtering
|
||||
):
|
||||
"""Test: si todos los canales son excluidos, resultado vacío."""
|
||||
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"
|
||||
mock_settings.include_text = []
|
||||
# Excluir texto común a todos
|
||||
mock_settings.exclude_text = ["a", "e", "i", "o", "u"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels_for_filtering)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Solo header, sin canales (todos tienen vocales)
|
||||
assert content == "#EXTM3U\n"
|
||||
|
||||
def test_write_m3u_include_startswith_not_contains(self, temp_dir):
|
||||
"""Test: include usa startswith, NO contains - verificación explícita."""
|
||||
channels = [
|
||||
{"name": "ABC News", "stream_id": 1, "stream_icon": "", "category_id": "1"},
|
||||
{
|
||||
"name": "News ABC",
|
||||
"stream_id": 2,
|
||||
"stream_icon": "",
|
||||
"category_id": "1",
|
||||
}, # Contiene ABC pero no empieza
|
||||
{
|
||||
"name": "ZABC Channel",
|
||||
"stream_id": 3,
|
||||
"stream_icon": "",
|
||||
"category_id": "1",
|
||||
}, # Contiene ABC pero no empieza
|
||||
]
|
||||
|
||||
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"
|
||||
mock_settings.include_text = ["ABC"]
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
manager._write_m3u(channels)
|
||||
|
||||
output_file = temp_dir / "playlist.m3u"
|
||||
content = output_file.read_text()
|
||||
|
||||
# Solo "ABC News" debe estar - empieza con ABC
|
||||
assert "ABC News" in content
|
||||
# "News ABC" y "ZABC Channel" NO deben estar
|
||||
assert "News ABC" not in content
|
||||
assert "ZABC Channel" not in content
|
||||
|
||||
# ========================================
|
||||
# Tests para fetch_and_write_epg - EPG Functionality
|
||||
# ========================================
|
||||
|
||||
def test_fetch_and_write_epg_success(self, temp_dir):
|
||||
"""Test: fetch_and_write_epg descarga y guarda EPG correctamente."""
|
||||
sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tv>
|
||||
<channel id="ch1">
|
||||
<display-name>ESPN Sports</display-name>
|
||||
</channel>
|
||||
<programme channel="ch1" start="20260203" stop="20260203">
|
||||
<title>Sports Event</title>
|
||||
</programme>
|
||||
</tv>'''
|
||||
|
||||
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.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
with patch("m3u_list_builder.playlist.requests.get") as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = sample_epg
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
manager.fetch_and_write_epg()
|
||||
|
||||
# Verify EPG file was written
|
||||
epg_file = temp_dir / "epg.xml"
|
||||
assert epg_file.exists()
|
||||
content = epg_file.read_text()
|
||||
assert "ESPN Sports" in content
|
||||
|
||||
def test_fetch_and_write_epg_correct_api_call(self, temp_dir):
|
||||
"""Test: fetch_and_write_epg llama a la API correctamente."""
|
||||
sample_epg = b'<?xml version="1.0"?><tv></tv>'
|
||||
|
||||
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.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
with patch("m3u_list_builder.playlist.requests.get") as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = sample_epg
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
manager.fetch_and_write_epg()
|
||||
|
||||
mock_get.assert_called_once_with(
|
||||
"http://test-iptv.com/xmltv.php",
|
||||
params={
|
||||
"username": "testuser",
|
||||
"password": "testpass",
|
||||
},
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
def test_fetch_and_write_epg_request_exception(self, caplog):
|
||||
"""Test: fetch_and_write_epg maneja RequestException sin crash."""
|
||||
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"
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
with patch("m3u_list_builder.playlist.requests.get") as mock_get:
|
||||
mock_get.side_effect = requests.RequestException("Connection failed")
|
||||
|
||||
manager.fetch_and_write_epg()
|
||||
|
||||
assert "Error de red al obtener EPG" in caplog.text
|
||||
|
||||
def test_fetch_and_write_epg_parse_error(self, caplog):
|
||||
"""Test: fetch_and_write_epg maneja error de parsing XML."""
|
||||
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"
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
with patch("m3u_list_builder.playlist.requests.get") as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = b"invalid xml <not closed"
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
manager.fetch_and_write_epg()
|
||||
|
||||
assert "Error parseando XML de EPG" in caplog.text
|
||||
|
||||
def test_fetch_and_write_epg_unexpected_exception(self, caplog):
|
||||
"""Test: fetch_and_write_epg maneja excepciones inesperadas."""
|
||||
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.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
with patch("m3u_list_builder.playlist.requests.get") as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = b'<?xml version="1.0"?><tv></tv>'
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with patch.object(
|
||||
manager, "_write_epg", side_effect=RuntimeError("Disk full")
|
||||
):
|
||||
manager.fetch_and_write_epg()
|
||||
|
||||
assert "Error inesperado actualizando EPG" in caplog.text
|
||||
|
||||
# ========================================
|
||||
# Tests para _filter_epg - Filtrado de EPG
|
||||
# ========================================
|
||||
|
||||
def test_filter_epg_include_text_filters_by_prefix(self):
|
||||
"""Test: _filter_epg filtra canales que EMPIEZAN con el texto."""
|
||||
sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tv>
|
||||
<channel id="ch1">
|
||||
<display-name>ESPN Sports</display-name>
|
||||
</channel>
|
||||
<channel id="ch2">
|
||||
<display-name>Sports Center ESPN</display-name>
|
||||
</channel>
|
||||
<channel id="ch3">
|
||||
<display-name>HBO Movies</display-name>
|
||||
</channel>
|
||||
<programme channel="ch1" start="20260203" stop="20260203">
|
||||
<title>Sports Event</title>
|
||||
</programme>
|
||||
<programme channel="ch2" start="20260203" stop="20260203">
|
||||
<title>Center Show</title>
|
||||
</programme>
|
||||
<programme channel="ch3" start="20260203" stop="20260203">
|
||||
<title>Movie</title>
|
||||
</programme>
|
||||
</tv>'''
|
||||
|
||||
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.include_text = ["ESPN"]
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
manager = PlaylistManager()
|
||||
result = manager._filter_epg(sample_epg)
|
||||
|
||||
# Convert to string for easier assertion
|
||||
result_str = ET.tostring(result, encoding="unicode")
|
||||
|
||||
# ESPN Sports starts with ESPN - should be included
|
||||
assert "ESPN Sports" in result_str
|
||||
# Sports Center ESPN does NOT start with ESPN - should be filtered
|
||||
assert "Sports Center ESPN" not in result_str
|
||||
# HBO Movies - should be filtered
|
||||
assert "HBO Movies" not in result_str
|
||||
# Programme for ch1 should remain
|
||||
assert "Sports Event" in result_str
|
||||
# Programmes for ch2 and ch3 should be removed
|
||||
assert "Center Show" not in result_str
|
||||
assert "Movie" not in result_str
|
||||
|
||||
def test_filter_epg_exclude_text_filters_by_contains(self):
|
||||
"""Test: _filter_epg excluye canales que CONTIENEN el texto."""
|
||||
sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tv>
|
||||
<channel id="ch1">
|
||||
<display-name>ESPN Sports</display-name>
|
||||
</channel>
|
||||
<channel id="ch2">
|
||||
<display-name>Adult Content</display-name>
|
||||
</channel>
|
||||
<programme channel="ch1" start="20260203" stop="20260203">
|
||||
<title>Sports Event</title>
|
||||
</programme>
|
||||
<programme channel="ch2" start="20260203" stop="20260203">
|
||||
<title>Adult Show</title>
|
||||
</programme>
|
||||
</tv>'''
|
||||
|
||||
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.include_text = []
|
||||
mock_settings.exclude_text = ["Adult"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
manager = PlaylistManager()
|
||||
result = manager._filter_epg(sample_epg)
|
||||
|
||||
result_str = ET.tostring(result, encoding="unicode")
|
||||
|
||||
# ESPN Sports - should remain
|
||||
assert "ESPN Sports" in result_str
|
||||
# Adult Content - should be filtered
|
||||
assert "Adult Content" not in result_str
|
||||
# Programme for ch1 should remain
|
||||
assert "Sports Event" in result_str
|
||||
# Programme for ch2 should be removed
|
||||
assert "Adult Show" not in result_str
|
||||
|
||||
def test_filter_epg_no_filters_keeps_all(self):
|
||||
"""Test: _filter_epg sin filtros mantiene todos los canales."""
|
||||
sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tv>
|
||||
<channel id="ch1">
|
||||
<display-name>Channel 1</display-name>
|
||||
</channel>
|
||||
<channel id="ch2">
|
||||
<display-name>Channel 2</display-name>
|
||||
</channel>
|
||||
</tv>'''
|
||||
|
||||
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.include_text = []
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
manager = PlaylistManager()
|
||||
result = manager._filter_epg(sample_epg)
|
||||
|
||||
result_str = ET.tostring(result, encoding="unicode")
|
||||
|
||||
assert "Channel 1" in result_str
|
||||
assert "Channel 2" in result_str
|
||||
|
||||
def test_filter_epg_include_and_exclude_combined(self):
|
||||
"""Test: _filter_epg aplica include y exclude juntos."""
|
||||
sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tv>
|
||||
<channel id="ch1">
|
||||
<display-name>ESPN Sports</display-name>
|
||||
</channel>
|
||||
<channel id="ch2">
|
||||
<display-name>ESPN Adult</display-name>
|
||||
</channel>
|
||||
<channel id="ch3">
|
||||
<display-name>HBO Movies</display-name>
|
||||
</channel>
|
||||
</tv>'''
|
||||
|
||||
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.include_text = ["ESPN"]
|
||||
mock_settings.exclude_text = ["Adult"]
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
manager = PlaylistManager()
|
||||
result = manager._filter_epg(sample_epg)
|
||||
|
||||
result_str = ET.tostring(result, encoding="unicode")
|
||||
|
||||
# ESPN Sports - starts with ESPN, no Adult - should remain
|
||||
assert "ESPN Sports" in result_str
|
||||
# ESPN Adult - starts with ESPN but contains Adult - should be filtered
|
||||
assert "ESPN Adult" not in result_str
|
||||
# HBO Movies - doesn't start with ESPN - should be filtered
|
||||
assert "HBO Movies" not in result_str
|
||||
|
||||
def test_filter_epg_case_insensitive(self):
|
||||
"""Test: _filter_epg es case-insensitive."""
|
||||
sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tv>
|
||||
<channel id="ch1">
|
||||
<display-name>ESPN Sports</display-name>
|
||||
</channel>
|
||||
</tv>'''
|
||||
|
||||
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.include_text = ["espn"] # lowercase
|
||||
mock_settings.exclude_text = []
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
manager = PlaylistManager()
|
||||
result = manager._filter_epg(sample_epg)
|
||||
|
||||
result_str = ET.tostring(result, encoding="unicode")
|
||||
|
||||
# Should still match despite case difference
|
||||
assert "ESPN Sports" in result_str
|
||||
|
||||
# ========================================
|
||||
# Tests para _write_epg
|
||||
# ========================================
|
||||
|
||||
def test_write_epg_creates_file(self, temp_dir):
|
||||
"""Test: _write_epg crea el archivo EPG correctamente."""
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
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"
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
root = ET.Element("tv")
|
||||
channel = ET.SubElement(root, "channel", id="ch1")
|
||||
name = ET.SubElement(channel, "display-name")
|
||||
name.text = "Test Channel"
|
||||
|
||||
manager._write_epg(root)
|
||||
|
||||
epg_file = temp_dir / "epg.xml"
|
||||
assert epg_file.exists()
|
||||
content = epg_file.read_text()
|
||||
assert "Test Channel" in content
|
||||
|
||||
def test_write_epg_atomic_replacement(self, temp_dir):
|
||||
"""Test: _write_epg reemplaza atómicamente."""
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
epg_file = temp_dir / "epg.xml"
|
||||
temp_file = temp_dir / "epg.xml.tmp"
|
||||
|
||||
# Create existing file
|
||||
epg_file.write_text("OLD CONTENT")
|
||||
|
||||
with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
|
||||
with patch("m3u_list_builder.playlist.settings") as mock_settings:
|
||||
mock_settings.host = "http://test-iptv.com"
|
||||
mock_settings.username = "testuser"
|
||||
mock_settings.password = "testpass"
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
root = ET.Element("tv")
|
||||
manager._write_epg(root)
|
||||
|
||||
# Temp file should not exist after
|
||||
assert not temp_file.exists()
|
||||
# Final file should have new content
|
||||
assert "OLD CONTENT" not in epg_file.read_text()
|
||||
|
||||
def test_fetch_and_generate_calls_epg(self, sample_channels):
|
||||
"""Test: fetch_and_generate también llama a fetch_and_write_epg."""
|
||||
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 = "test_playlist.m3u"
|
||||
|
||||
from m3u_list_builder.playlist import PlaylistManager
|
||||
|
||||
manager = PlaylistManager()
|
||||
|
||||
with patch("m3u_list_builder.playlist.requests.get") as mock_get:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = sample_channels
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with patch.object(manager, "_write_m3u"):
|
||||
with patch.object(
|
||||
manager, "fetch_and_write_epg"
|
||||
) as mock_epg:
|
||||
manager.fetch_and_generate()
|
||||
|
||||
mock_epg.assert_called_once()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user