Compare commits

..

12 Commits

Author SHA1 Message Date
df17eaaf12 Merge pull request 'Dev' (#12) from Dev into main
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 27s
CI/CD Pipeline / publish-container (push) Successful in 6m6s
Reviewed-on: #12
2026-02-03 16:56:16 +00:00
fb7eff451d Merge pull request 'epg_func' (#11) from epg_func into Dev
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 27s
CI/CD Pipeline / test-and-lint (pull_request) Successful in 27s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / publish-container (pull_request) Has been skipped
Reviewed-on: #11
2026-02-03 15:37:46 +00:00
9e8f3bcb47 fix: update version to 0.3.2 in pyproject.toml and enhance README with EPG and filtering details
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 27s
CI/CD Pipeline / publish-container (push) Has been skipped
CI/CD Pipeline / test-and-lint (pull_request) Successful in 27s
CI/CD Pipeline / publish-container (pull_request) Has been skipped
2026-02-03 15:35:48 +00:00
34420e6758 feat: add tests for _write_m3u handling of epg_channel_id presence and absence 2026-02-03 15:31:48 +00:00
545958028d feat: implement EPG fetching and filtering functionality in PlaylistManager
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 39s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-03 15:28:49 +00:00
e62c243542 Merge pull request 'main' (#10) from main into Dev
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 24s
CI/CD Pipeline / publish-container (push) Has been skipped
Reviewed-on: #10
2026-02-02 16:46:52 +00:00
6c357dd977 Merge pull request 'Dev' (#9) from Dev into main
All checks were successful
CI/CD Pipeline / test-and-lint (pull_request) Successful in 27s
CI/CD Pipeline / publish-container (pull_request) Has been skipped
CI/CD Pipeline / test-and-lint (push) Successful in 26s
CI/CD Pipeline / publish-container (push) Successful in 5m14s
Reviewed-on: #9
2026-02-02 16:45:33 +00:00
7733c7e2ac Merge pull request 'Actualizar el README para acomodar la nueva funcionalidad' (#8) from unai-patch-5 into main
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 25s
CI/CD Pipeline / publish-container (push) Has been skipped
Reviewed-on: #8
2026-02-02 16:40:37 +00:00
c1d9b6bf54 feat: add include_text and exclude_text filters to settings and playlist tests
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 25s
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-02 16:28:51 +00:00
757ed91aae Linting, line too long 2026-02-02 16:26:52 +00:00
1cf174e896 feat: add inclusion and exclusion filters for channel names in playlist generation
Some checks failed
CI/CD Pipeline / test-and-lint (push) Failing after 22s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-02 16:23:45 +00:00
ab9ccd16aa feat: add .dockerignore to exclude unnecessary files from Docker builds
All checks were successful
CI/CD Pipeline / test-and-lint (push) Successful in 44s
CI/CD Pipeline / publish-container (push) Has been skipped
2026-02-02 16:17:38 +00:00
8 changed files with 1257 additions and 22 deletions

30
.dockerignore Normal file
View 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

View File

@ -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
@ -41,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
@ -74,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
@ -87,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) │ │

View File

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

View File

@ -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()

View File

@ -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)

View File

@ -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()

View File

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

View File

@ -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()