117 lines
4.6 KiB
Python
117 lines
4.6 KiB
Python
"""
|
|
Recolección de partes de un ZIP multipart, sin importar la caja de la extensión.
|
|
|
|
En este dominio los respaldos llegan con extensión en MAYÚSCULAS con frecuencia (los propios
|
|
tests del panel usan `GENERICA-TEST.ZIP`). El bug que esto fija: `_collect_zip_paths` hacía
|
|
`path.stem.split(".zip")[0]`, que con `EMPRESA.ZIP.001` dejaba `base_name="EMPRESA.ZIP"` y
|
|
armaba el glob `EMPRESA.ZIP.zip.*`, que no encuentra nada. Devolvía lista vacía, y como
|
|
`_move_zip_to_processed` y `_move_zip_to_failed` iteran sobre ese resultado, **las partes nunca
|
|
salían de Entrada**: se acumulaban ahí mezcladas con los pendientes.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
class _FakeExtractor:
|
|
"""Espeja is_multipart de SevenZipExtractor: reconoce .zip.NNN sin importar la caja."""
|
|
|
|
@staticmethod
|
|
def is_multipart(zip_path: str) -> bool:
|
|
path = Path(zip_path)
|
|
# .zip.001 -> suffix ".001", stem "algo.zip"
|
|
return path.stem.lower().endswith(".zip") and len(path.suffix) == 4
|
|
|
|
|
|
def _collect(source_path: str) -> list[str]:
|
|
"""
|
|
Copia de la lógica de RestoreWorker._collect_zip_paths, aislada para poder probarla sin
|
|
arrastrar PySide6 ni pyodbc. Si la implementación cambia, este test debe cambiar con ella.
|
|
"""
|
|
path = Path(source_path)
|
|
if not _FakeExtractor.is_multipart(str(path)):
|
|
return [str(path)]
|
|
|
|
stem = path.stem
|
|
base_name = stem[:-4] if stem.lower().endswith(".zip") else stem
|
|
prefix = f"{base_name}.zip.".lower()
|
|
|
|
parts = sorted(
|
|
(item for item in path.parent.iterdir() if item.name.lower().startswith(prefix)),
|
|
key=lambda item: item.name.lower(),
|
|
)
|
|
if parts:
|
|
return [str(item) for item in parts]
|
|
return [str(path)]
|
|
|
|
|
|
@pytest.mark.parametrize("ext", ["zip", "ZIP", "Zip"])
|
|
def test_recolecta_todas_las_partes_sin_importar_la_caja(tmp_path: Path, ext: str):
|
|
"""Las tres partes se recolectan igual con la extensión en minúsculas, MAYÚSCULAS o mixta."""
|
|
names = [f"EMPRESA.{ext}.001", f"EMPRESA.{ext}.002", f"EMPRESA.{ext}.003"]
|
|
for name in names:
|
|
(tmp_path / name).write_bytes(b"x")
|
|
|
|
collected = _collect(str(tmp_path / names[0]))
|
|
|
|
assert [Path(p).name for p in collected] == names, (
|
|
f"con extensión .{ext} se recolectaron {len(collected)} de {len(names)} partes"
|
|
)
|
|
|
|
|
|
def test_orden_estable_entre_partes(tmp_path: Path):
|
|
"""El orden importa: 7-Zip necesita la .001 primero para reensamblar."""
|
|
for i in (3, 1, 10, 2):
|
|
(tmp_path / f"BASE.ZIP.{i:03d}").write_bytes(b"x")
|
|
|
|
collected = [Path(p).name for p in _collect(str(tmp_path / "BASE.ZIP.001"))]
|
|
assert collected == ["BASE.ZIP.001", "BASE.ZIP.002", "BASE.ZIP.003", "BASE.ZIP.010"]
|
|
|
|
|
|
def test_no_mezcla_partes_de_otro_respaldo(tmp_path: Path):
|
|
"""Dos multipart en la misma carpeta no deben contaminarse entre sí."""
|
|
for name in ["ALFA.ZIP.001", "ALFA.ZIP.002", "OMEGA.ZIP.001", "OMEGA.zip.002"]:
|
|
(tmp_path / name).write_bytes(b"x")
|
|
|
|
alfa = [Path(p).name for p in _collect(str(tmp_path / "ALFA.ZIP.001"))]
|
|
assert alfa == ["ALFA.ZIP.001", "ALFA.ZIP.002"]
|
|
|
|
# OMEGA tiene sus dos partes con distinta caja: aun así deben salir las dos.
|
|
omega = [Path(p).name for p in _collect(str(tmp_path / "OMEGA.ZIP.001"))]
|
|
assert sorted(omega) == ["OMEGA.ZIP.001", "OMEGA.zip.002"]
|
|
|
|
|
|
def test_zip_simple_devuelve_solo_ese_archivo(tmp_path: Path):
|
|
simple = tmp_path / "UNICO.ZIP"
|
|
simple.write_bytes(b"x")
|
|
assert _collect(str(simple)) == [str(simple)]
|
|
|
|
|
|
def test_multipart_sin_partes_localizadas_devuelve_el_original(tmp_path: Path):
|
|
"""
|
|
Si no se localizan las partes, se devuelve el archivo original en lugar de lista vacía:
|
|
mover una sola parte es mejor que dejarla atorada en Entrada indefinidamente.
|
|
"""
|
|
huerfana = tmp_path / "SOLA.ZIP.007"
|
|
huerfana.write_bytes(b"x")
|
|
# Es multipart por el nombre, y su propia parte sí se encuentra.
|
|
assert _collect(str(huerfana)) == [str(huerfana)]
|
|
|
|
|
|
def test_la_implementacion_real_no_usa_glob_en_minusculas():
|
|
"""
|
|
Tripwire sobre el código real: `glob` distingue mayúsculas en Linux, así que un patrón
|
|
en minúsculas nunca encontraría `.ZIP.001`. La implementación debe filtrar iterdir()
|
|
comparando en minúsculas.
|
|
"""
|
|
source = (
|
|
Path(__file__).resolve().parent.parent / "app" / "engine" / "restore_worker.py"
|
|
).read_text(encoding="utf-8")
|
|
start = source.index("def _collect_zip_paths")
|
|
body = source[start : start + 2000]
|
|
|
|
assert 'glob(f"{base_name}.zip.*")' not in body, "glob en minúsculas: no halla .ZIP en Linux"
|
|
assert "iterdir()" in body
|
|
assert ".lower()" in body
|