0819f2da89
Munka no podía reconvertir archivos que antes sí. Auditado sobre la carpeta de Andrea Galeano: - Los BRAW nunca habían salido: 87 clips (496 GB), tres tandas por el guion de Resolve y ni un clip convertido. Ahora los lee el decodificador de Blackmagic que instala Resolve (braw.py, en un proceso aparte), sin abrir Resolve, y ffmpeg los codifica como cualquier otro clip. Probado con clips del NAS: 15,6 GB → 394 MB con sus 7.004 cuadros, VMAF 96,0 contra el BRAW revelado sin pérdida. Las conversiones de BRAW que fallaron por Resolve pasan solas al lector. - Lo que «antes podía» eran reconversiones de lo ya convertido: ahorraban un 5 % con pérdida. Siguen sin ofrecerse, y lo que ya viene comprimido de cámara solo se ofrece si compensa. - La tarjeta codifica con «uhq»: entre un 19 y un 55 % menos de peso con la misma calidad medida, siempre H.265 de 10 bits. - Una conversión que no aligera (más del 80 % del clip, o reconversión de una conversión) no reemplaza al original: queda «no compensa» y lo que salió va a _Para borrar\descartados. De paso: los MP4 guardan las etiquetas de la cámara; un intento que no pasó la verificación se puede reintentar; un H.264 de más de 4.096 px de lado va por CPU; el motivo de un fallo se lee entero; el guion de Resolve llega con tildes y borra su ficha. Pruebas: prueba_orden.py 33 (una con un BRAW de verdad), probar.py 19, prueba_funcional.py 30, prueba_ventana.py 18 desde el código y 17 con el paquete. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
644 líneas
36 KiB
Python
644 líneas
36 KiB
Python
# -*- coding: utf-8 -*-
|
|
r"""
|
|
Pruebas automáticas de OrganiLAB. Se corren con: py -3.12 probar.py
|
|
|
|
- reglas: detección de cámara, mes, año, basura y proxies
|
|
- firma: firmar/verificar y rechazo de manifiestos alterados o sin firmar
|
|
- servidor (extremo a extremo): copia el programa a una carpeta temporal, lo arranca en un puerto libre,
|
|
registra una carpeta de ensayo, la escanea, calcula el plan, lo aplica, lo deshace, y la desconecta.
|
|
No toca datos\ ni el inventario real: todo ocurre en la carpeta temporal.
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import shutil
|
|
import socket
|
|
import tempfile
|
|
import unittest
|
|
import subprocess
|
|
import urllib.parse
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
AQUI = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, AQUI)
|
|
import reglas as R # noqa: E402
|
|
import firma # noqa: E402
|
|
|
|
|
|
class Reglas(unittest.TestCase):
|
|
def test_firma_camara(self):
|
|
self.assertEqual(R.firma_archivo("C0001.MP4"), "SONY")
|
|
self.assertEqual(R.firma_archivo("C0001M01.XML"), "SONY")
|
|
self.assertEqual(R.firma_archivo("20260301_labstream0001.mp4"), "SONY")
|
|
self.assertIsNone(R.firma_archivo("informe final.docx"))
|
|
|
|
def test_mes(self):
|
|
self.assertEqual(R.detectar_mes("09- Septiembre"), (9, False))
|
|
self.assertEqual(R.detectar_mes("02-marzo"), (3, True)) # gana el nombre, con conflicto
|
|
self.assertEqual(R.detectar_mes("MAYO"), (5, False))
|
|
self.assertEqual(R.detectar_mes("3"), (3, True)) # número solo: confianza media
|
|
self.assertEqual(R.detectar_mes("13"), (None, False))
|
|
self.assertEqual(R.detectar_mes("Brigard"), (None, False))
|
|
|
|
def test_anio(self):
|
|
self.assertEqual(R.detectar_anio("2026"), 2026)
|
|
self.assertIsNone(R.detectar_anio("2026 - agosto"))
|
|
|
|
def test_app_proyecto(self):
|
|
self.assertEqual(R.app_proyecto(".prproj"), "Premiere")
|
|
self.assertEqual(R.app_proyecto(".AEP"), "After Effects")
|
|
self.assertEqual(R.app_proyecto(".drp"), "DaVinci Resolve")
|
|
self.assertEqual(R.app_proyecto(".psd"), "Photoshop")
|
|
self.assertIsNone(R.app_proyecto(".mp4"))
|
|
# la plantilla de proyectos separa por programa
|
|
self.assertIn("{app}", R.PLANTILLAS["contenido"]["PROYECTOS"])
|
|
|
|
def test_dia(self):
|
|
self.assertEqual(R.nombre_dia("2026-05-15"), "15- Viernes")
|
|
self.assertEqual(R.nombre_dia("2025-12-17"), "17- Miércoles")
|
|
self.assertEqual(R.nombre_dia("nada"), "")
|
|
# la plantilla de grabación lleva el día, que se rellena solo cuando hace falta
|
|
self.assertIn("{dia}", R.PLANTILLAS["contenido"]["GRABACION"])
|
|
|
|
def test_basura_y_proxies(self):
|
|
self.assertTrue(R.es_basura_archivo("Thumbs.db"))
|
|
self.assertTrue(R.es_basura_archivo(".DS_Store"))
|
|
self.assertFalse(R.es_basura_archivo("C0001.MP4"))
|
|
self.assertTrue(R.es_proxy_dir("Proxy"))
|
|
self.assertFalse(R.es_proxy_dir("Entregas"))
|
|
|
|
|
|
class Config(unittest.TestCase):
|
|
def test_config_roto_se_aparta(self):
|
|
"""Un config.json ilegible no se pisa en silencio: queda una copia .roto-* y se parte de cero."""
|
|
tmp = tempfile.mkdtemp(prefix="organilab-cfg-")
|
|
try:
|
|
ruta = os.path.join(tmp, "config.json")
|
|
with open(ruta, "w", encoding="utf-8") as fh:
|
|
fh.write('{"raices": [ esto no es json')
|
|
src = open(os.path.join(AQUI, "organizador.py"), encoding="utf-8").read()
|
|
ini = src.index("def cargar_json(")
|
|
fin = src.index("def guardar_json(")
|
|
ns = {"json": json, "os": os, "shutil": shutil, "time": time}
|
|
exec(src[ini:fin], ns)
|
|
out = ns["cargar_json"](ruta, {"raices": [], "a": 1})
|
|
self.assertEqual(out, {"raices": [], "a": 1})
|
|
self.assertTrue([f for f in os.listdir(tmp) if f.startswith("config.json.roto-")], "no dejó copia del roto")
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
@unittest.skipUnless(os.name == "nt", "la ventana y los accesos directos son de Windows")
|
|
class Windows(unittest.TestCase):
|
|
def test_tablas_de_la_ventana(self):
|
|
"""Todos los identificadores de interfaz de ventana.py están bien escritos, y preguntar si hay
|
|
WebView2 no revienta en ningún equipo (la respuesta depende del equipo)."""
|
|
import ventana
|
|
ids = [v for k, v in vars(ventana).items() if k.startswith(("IID_", "CLSID_", "FMT_")) and isinstance(v, str)]
|
|
self.assertGreater(len(ids), 20)
|
|
for g in ids:
|
|
self.assertEqual(len(bytes(ventana._guid(g))), 16)
|
|
ok, detalle = ventana.disponible(AQUI)
|
|
self.assertIsInstance(ok, bool)
|
|
self.assertTrue(detalle)
|
|
self.assertFalse(ventana.traer_al_frente(123456789)) # nadie abierto con esa marca
|
|
|
|
def test_acceso_directo(self):
|
|
"""El acceso directo que se escribe dice lo que se le pidió, identidad de OrganiLAB incluida: sin
|
|
ella, la barra de tareas trataría el acceso anclado y la ventana abierta como dos programas."""
|
|
import ventana
|
|
import instalar
|
|
tmp = tempfile.mkdtemp(prefix="organilab-lnk-")
|
|
try:
|
|
exe = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "notepad.exe")
|
|
ruta = instalar.acceso_directo(os.path.join(tmp, "OrganiLAB.lnk"), exe, '"una carpeta\\organizador.py"', tmp,
|
|
os.path.join(AQUI, "static", "organilab.ico"))
|
|
a = instalar.leer_acceso(ruta)
|
|
self.assertEqual(os.path.normcase(a["exe"]), os.path.normcase(exe))
|
|
self.assertEqual(a["args"], '"una carpeta\\organizador.py"')
|
|
self.assertEqual(os.path.normcase(a["carpeta"]), os.path.normcase(tmp))
|
|
self.assertTrue(a["icono"].lower().endswith("organilab.ico"))
|
|
self.assertEqual(a["identidad"], ventana.IDENTIDAD)
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_instalar_y_quitar(self):
|
|
"""Dar de alta el programa en Windows y quitarlo, de punta a punta pero sin tocar el menú Inicio ni el
|
|
Escritorio de verdad: los accesos van a una carpeta temporal y la línea de Aplicaciones a una clave de ensayo."""
|
|
import winreg
|
|
import instalar
|
|
tmp = tempfile.mkdtemp(prefix="organilab-alta-")
|
|
base = os.path.join(tmp, "programa")
|
|
os.makedirs(os.path.join(base, "static"))
|
|
inicio, escritorio = os.path.join(tmp, "inicio", "OrganiLAB.lnk"), os.path.join(tmp, "escritorio", "OrganiLAB.lnk")
|
|
os.makedirs(os.path.dirname(inicio))
|
|
os.makedirs(os.path.dirname(escritorio))
|
|
viejos = instalar._accesos, instalar.CLAVE
|
|
instalar._accesos, instalar.CLAVE = (lambda: (inicio, escritorio)), r"Software\OrganiLAB-ensayo\Alta"
|
|
try:
|
|
self.assertIsNone(instalar.instalado())
|
|
r = instalar.instalar(base, "3.0.0")
|
|
self.assertEqual(sorted(r["accesos"]), sorted([inicio, escritorio]))
|
|
self.assertTrue(instalar.instalado_aqui(base))
|
|
self.assertFalse(instalar.instalado_aqui(tmp))
|
|
self.assertEqual(os.path.normcase(instalar.leer_acceso(inicio)["carpeta"]), os.path.normcase(base))
|
|
# tras una actualización: sube la versión que enseña Windows, y no repone el acceso que alguien quitó
|
|
self.assertFalse(instalar.al_dia(base, "3.0.0"))
|
|
os.remove(escritorio)
|
|
self.assertTrue(instalar.al_dia(base, "3.0.1"))
|
|
self.assertEqual(instalar.instalado()["version"], "3.0.1")
|
|
self.assertFalse(os.path.exists(escritorio))
|
|
# un acceso directo que no se deja reescribir (pasó con el del Escritorio en OneDrive) no impide
|
|
# apuntar la versión nueva; y uno que ya abre lo que debe ni se toca
|
|
open(os.path.join(base, "OrganizadorLabstream.exe"), "wb").close() # ahora el programa se abre con otra cosa
|
|
de_verdad, intentos = instalar.acceso_directo, []
|
|
instalar.acceso_directo = lambda *a, **k: (intentos.append(a[0]), (_ for _ in ()).throw(OSError("ocupado")))[1]
|
|
try:
|
|
self.assertTrue(instalar.al_dia(base, "3.0.2"))
|
|
finally:
|
|
instalar.acceso_directo = de_verdad
|
|
self.assertEqual(intentos, [inicio])
|
|
self.assertEqual(instalar.instalado()["version"], "3.0.2")
|
|
self.assertTrue(instalar.al_dia(base, "3.0.2")) # ahora sí se deja: lo pone al día
|
|
self.assertTrue(instalar._misma(instalar.leer_acceso(inicio)["exe"], os.path.join(base, "OrganizadorLabstream.exe")))
|
|
self.assertFalse(instalar.al_dia(base, "3.0.2")) # y ya no queda nada que cambiar
|
|
os.remove(os.path.join(base, "OrganizadorLabstream.exe"))
|
|
instalar.al_dia(base, "3.0.2")
|
|
# quitarlo desde otra copia del programa no toca lo de esta; desde esta, sí
|
|
self.assertEqual(instalar.desinstalar(tmp)["accesos"], [])
|
|
self.assertTrue(instalar.instalado_aqui(base))
|
|
# desinstalar de verdad quita también la tarea del orden automático; contra la clave de ensayo
|
|
# no debe ni intentarlo: aquí se comprueba que no llama al Programador de tareas
|
|
llamadas, correr = [], instalar.subprocess.run
|
|
instalar.subprocess.run = lambda *a, **k: llamadas.append(a)
|
|
try:
|
|
self.assertEqual(instalar.desinstalar(base)["accesos"], [inicio])
|
|
finally:
|
|
instalar.subprocess.run = correr
|
|
self.assertEqual(llamadas, [])
|
|
self.assertIsNone(instalar.instalado())
|
|
finally:
|
|
for clave in (instalar.CLAVE, r"Software\OrganiLAB-ensayo"):
|
|
try:
|
|
winreg.DeleteKey(winreg.HKEY_CURRENT_USER, clave)
|
|
except OSError:
|
|
pass
|
|
instalar._accesos, instalar.CLAVE = viejos
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_sin_ventana_y_con_el_puerto_ocupado(self):
|
|
"""Dos imprevistos a la vez. El puerto de siempre lo tiene otro programa: se toma uno libre en vez de
|
|
abrir la página de ese otro. Y este Windows no puede abrir la ventana (aquí, porque falta el cargador
|
|
de WebView2): la interfaz va al navegador y el programa se va solo cuando esa página deja de preguntar."""
|
|
tmp = tempfile.mkdtemp(prefix="organilab-imprevistos-")
|
|
# quien ocupa el puerto es un servidor como el de otro OrganiLAB: pide «reutilizar dirección», y en
|
|
# Windows eso dejaba que un segundo programa escuchara en el mismo puerto sin que nada fallara
|
|
import http.server
|
|
tapon = http.server.ThreadingHTTPServer(("127.0.0.1", 0), http.server.BaseHTTPRequestHandler)
|
|
ocupado = tapon.server_address[1]
|
|
try:
|
|
for f in ("organizador.py", "reglas.py", "asistente.py", "actualizador.py", "resolve_bridge.py", "braw.py", "firma.py", "ventana.py", "instalar.py"):
|
|
shutil.copy(os.path.join(AQUI, f), tmp)
|
|
shutil.copytree(os.path.join(AQUI, "static"), os.path.join(tmp, "static"))
|
|
guion = ("import sys; sys.argv = ['organizador.py']; import organizador as O; "
|
|
f"O.PUERTO = {ocupado}; O.SILENCIO_NAVEGADOR = 2; "
|
|
"O.webbrowser.open = lambda u: print('ABRE', u, flush=True); O.main(); print('FIN', O.PUERTO, flush=True)")
|
|
env = {k: v for k, v in os.environ.items() if k != "ORGANIZADOR_PUERTO"}
|
|
env["PYTHONIOENCODING"] = "utf-8"
|
|
p = subprocess.run([sys.executable, "-c", guion], cwd=tmp, env=env, capture_output=True, text=True,
|
|
encoding="utf-8", errors="replace", timeout=90)
|
|
self.assertEqual(p.returncode, 0, p.stderr[-600:])
|
|
abre = [l for l in p.stdout.splitlines() if l.startswith("ABRE ")]
|
|
self.assertEqual(len(abre), 1, p.stdout[-600:])
|
|
self.assertNotIn(f":{ocupado}/", abre[0])
|
|
self.assertIn("Sin ventana propia", p.stdout)
|
|
self.assertTrue(p.stdout.strip().splitlines()[-1].startswith("FIN "))
|
|
self.assertFalse(os.path.exists(os.path.join(tmp, "datos", "sesion.json")))
|
|
finally:
|
|
tapon.server_close()
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_instalarse_en_su_sitio(self):
|
|
"""Instalar un paquete lo lleva a su sitio definitivo con sus datos, sin borrar nada del origen. No viaja
|
|
lo que es de la sesión abierta ni lo que se rehace solo, y la copia de origen queda sabiendo adónde se fue."""
|
|
import sqlite3
|
|
import winreg
|
|
import instalar
|
|
tmp = tempfile.mkdtemp(prefix="organilab-mudanza-")
|
|
base, destino = os.path.join(tmp, "Descargas", "OrganizadorLabstream"), os.path.join(tmp, "Programs", "OrganiLAB")
|
|
for rel in ("OrganizadorLabstream.exe", "organizador.py", "static/index.html", "static/organilab.ico", "ffmpeg/ffmpeg.exe",
|
|
"datos/config.json", "datos/bitacora.jsonl", "datos/sesion.json", "datos/ventana/EBWebView/cache.bin",
|
|
"actualizacion/nuevo.zip", "__pycache__/organizador.cpython-312.pyc"):
|
|
p = os.path.join(base, rel)
|
|
os.makedirs(os.path.dirname(p), exist_ok=True)
|
|
with open(p, "wb") as fh:
|
|
fh.write(rel.encode() * 50)
|
|
con = sqlite3.connect(os.path.join(base, "datos", "inventario.sqlite"))
|
|
con.execute("CREATE TABLE archivos (ruta TEXT)")
|
|
con.execute("INSERT INTO archivos VALUES ('V:/Cliente/C0001.MP4')")
|
|
con.commit()
|
|
con.close()
|
|
inicio, escritorio = os.path.join(tmp, "inicio", "OrganiLAB.lnk"), os.path.join(tmp, "escritorio", "OrganiLAB.lnk")
|
|
os.makedirs(os.path.dirname(inicio))
|
|
os.makedirs(os.path.dirname(escritorio))
|
|
viejos = instalar._accesos, instalar.CLAVE, instalar.sitio_definitivo
|
|
instalar._accesos, instalar.CLAVE, instalar.sitio_definitivo = (lambda: (inicio, escritorio)), r"Software\OrganiLAB-ensayo\Mudanza", (lambda: destino)
|
|
try:
|
|
self.assertTrue(instalar.hay_que_mudar(base))
|
|
self.assertEqual(instalar.mudado(base), "")
|
|
pasos = []
|
|
r = instalar.poner(base, "3.1.0", avance=pasos.append)
|
|
self.assertTrue(r["mudado"])
|
|
self.assertTrue(instalar._misma(r["base"], destino))
|
|
self.assertTrue(pasos and 0 < pasos[-1] <= 1)
|
|
for rel in ("OrganizadorLabstream.exe", "organizador.py", "static/index.html", "ffmpeg/ffmpeg.exe", "datos/config.json", "datos/bitacora.jsonl"):
|
|
self.assertTrue(os.path.exists(os.path.join(destino, rel)), rel)
|
|
for rel in ("datos/sesion.json", "datos/ventana", "actualizacion", "__pycache__", instalar.MARCA_MUDANZA):
|
|
self.assertFalse(os.path.exists(os.path.join(destino, rel)), rel)
|
|
con = sqlite3.connect(os.path.join(destino, "datos", "inventario.sqlite"))
|
|
self.assertEqual(con.execute("SELECT ruta FROM archivos").fetchall(), [("V:/Cliente/C0001.MP4",)])
|
|
con.close()
|
|
# del origen no se ha ido nada, y ya sabe adónde se instaló
|
|
for rel in ("OrganizadorLabstream.exe", "ffmpeg/ffmpeg.exe", "datos/sesion.json", "datos/inventario.sqlite"):
|
|
self.assertTrue(os.path.exists(os.path.join(base, rel)), rel)
|
|
self.assertTrue(instalar._misma(os.path.dirname(instalar.mudado(base)), destino))
|
|
# dado de alta desde su sitio, no desde Descargas
|
|
self.assertTrue(instalar.instalado_aqui(destino))
|
|
self.assertTrue(instalar._misma(instalar.leer_acceso(inicio)["exe"], os.path.join(destino, "OrganizadorLabstream.exe")))
|
|
# instalar otra vez desde el origen avisa de que ya hay uno; desde su sitio, es solo darlo de alta
|
|
with self.assertRaises(instalar.YaInstalado):
|
|
instalar.poner(base, "3.1.0")
|
|
self.assertFalse(instalar.poner(destino, "3.1.0")["mudado"])
|
|
# y un destino ocupado no se pisa nunca
|
|
instalar.desinstalar(destino)
|
|
with self.assertRaises(FileExistsError):
|
|
instalar.mudar(base)
|
|
self.assertTrue(os.path.exists(os.path.join(destino, "organizador.py")))
|
|
finally:
|
|
for clave in (instalar.CLAVE, r"Software\OrganiLAB-ensayo"):
|
|
try:
|
|
winreg.DeleteKey(winreg.HKEY_CURRENT_USER, clave)
|
|
except OSError:
|
|
pass
|
|
instalar._accesos, instalar.CLAVE, instalar.sitio_definitivo = viejos
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_orden_de_arranque(self):
|
|
"""Desde el código el programa se abre con el Python sin consola; en el paquete, con su propio .exe."""
|
|
import instalar
|
|
tmp = tempfile.mkdtemp(prefix="organilab-orden-")
|
|
try:
|
|
exe, args = instalar.orden(tmp)
|
|
self.assertTrue(exe.lower().endswith(("pythonw.exe", "python.exe")))
|
|
self.assertEqual(args, '"%s"' % os.path.join(tmp, "organizador.py"))
|
|
open(os.path.join(tmp, "OrganizadorLabstream.exe"), "wb").close()
|
|
self.assertEqual(instalar.orden(tmp), (os.path.join(tmp, "OrganizadorLabstream.exe"), ""))
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_arranque_no_estorba(self):
|
|
"""arranque.py es el sitecustomize del paquete. Fuera de un paquete (este Python no vive junto a un
|
|
organizador.py) y usado como intérprete normal, no hace nada: ni lanza el programa ni corta el proceso."""
|
|
p = subprocess.run([sys.executable, "-c", "import arranque; print('sigue')"], cwd=AQUI, capture_output=True, text=True, timeout=60)
|
|
self.assertEqual(p.stdout.strip(), "sigue")
|
|
|
|
|
|
class Manual(unittest.TestCase):
|
|
def test_manual_al_dia(self):
|
|
"""El manual vive dentro de la interfaz, con un apartado por cada pestaña del programa: si se añade una
|
|
vista y no se documenta, esto falla."""
|
|
import re
|
|
with open(os.path.join(AQUI, "static", "index.html"), encoding="utf-8") as fh:
|
|
html = fh.read()
|
|
vistas = re.findall(r'<a data-t="([a-z]+)"[^>]*>([^<]+)</a>', html[:html.index("</nav>")])
|
|
manual = html[html.index('<section id="t-manual">'):html.index("<!-- ============ AJUSTES")]
|
|
apartados = re.findall(r"<h3>([^<]+)</h3>", manual)
|
|
self.assertGreaterEqual(len(vistas), 8)
|
|
for clave, etiqueta in vistas:
|
|
if clave != "manual":
|
|
self.assertIn(etiqueta.replace("⚙", "").strip(), apartados)
|
|
self.assertGreater(manual.count("<dt>"), 35)
|
|
self.assertEqual(manual.count("<dt>"), manual.count("<dd>"))
|
|
|
|
|
|
class Firma(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmp = tempfile.mkdtemp(prefix="organilab-firma-")
|
|
cls.pub = firma.generar(os.path.join(cls.tmp, "clave.json"))
|
|
cls.priv = firma.cargar_privada(os.path.join(cls.tmp, "clave.json"))
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
shutil.rmtree(cls.tmp, ignore_errors=True)
|
|
|
|
def test_firmar_verificar(self):
|
|
f = firma.firmar(b"hola", self.priv)
|
|
self.assertTrue(firma.verificar(b"hola", f, self.pub))
|
|
self.assertFalse(firma.verificar(b"holA", f, self.pub))
|
|
self.assertFalse(firma.verificar(b"hola", f[:-2] + "00", self.pub))
|
|
self.assertFalse(firma.verificar(b"hola", "zz", self.pub))
|
|
|
|
def test_manifiesto(self):
|
|
import actualizador
|
|
m = {"version": "9.9.9", "app": {"archivo": "OrganizadorLabstream-app-9.9.9.zip", "sha256": "a" * 64, "size": 10, "url": "/api/herramientas/x.zip"}}
|
|
url = "https://os.labstreamsas.com/herramientas/organizador.json"
|
|
d, motivo = actualizador._validar(m, url)
|
|
self.assertIsNone(d)
|
|
self.assertIn("firmado", motivo)
|
|
m["firma"] = {"valor": firma.firmar(firma.texto_manifiesto(m), self.priv)}
|
|
# firmado con OTRA clave (la de prueba): tampoco vale para la clave real
|
|
self.assertIsNone(actualizador._validar(m, url)[0])
|
|
viejo = actualizador.CLAVE_PUBLICA
|
|
actualizador.CLAVE_PUBLICA = self.pub
|
|
try:
|
|
self.assertIsNotNone(actualizador._validar(m, url)[0])
|
|
m["app"]["sha256"] = "b" * 64
|
|
self.assertIsNone(actualizador._validar(m, url)[0])
|
|
finally:
|
|
actualizador.CLAVE_PUBLICA = viejo
|
|
|
|
|
|
def _puerto_libre():
|
|
s = socket.socket()
|
|
s.bind(("127.0.0.1", 0))
|
|
p = s.getsockname()[1]
|
|
s.close()
|
|
return p
|
|
|
|
|
|
class Servidor(unittest.TestCase):
|
|
"""Extremo a extremo sobre una copia temporal del programa."""
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmp = tempfile.mkdtemp(prefix="organilab-e2e-")
|
|
cls.app = os.path.join(cls.tmp, "app")
|
|
os.makedirs(cls.app)
|
|
for f in ("organizador.py", "reglas.py", "asistente.py", "actualizador.py", "resolve_bridge.py", "braw.py", "firma.py", "ventana.py", "instalar.py"):
|
|
shutil.copy(os.path.join(AQUI, f), cls.app)
|
|
shutil.copytree(os.path.join(AQUI, "static"), os.path.join(cls.app, "static"))
|
|
# carpeta de ensayo: cámara Sony, basura y un proxy
|
|
cls.raiz = os.path.join(cls.tmp, "Cliente Prueba")
|
|
for rel, tam in (("Rodaje/C0001.MP4", 3_000_000), ("Rodaje/C0001M01.XML", 500), ("Rodaje/Thumbs.db", 100),
|
|
("Rodaje/Proxy/C0001.MOV", 200_000), ("Entregas/Video final.mp4", 800_000), ("notas.txt", 50)):
|
|
p = os.path.join(cls.raiz, rel)
|
|
os.makedirs(os.path.dirname(p), exist_ok=True)
|
|
with open(p, "wb") as fh:
|
|
fh.write(os.urandom(min(tam, 4096)) * max(1, tam // 4096))
|
|
cls.puerto = _puerto_libre()
|
|
env = dict(os.environ, ORGANIZADOR_PUERTO=str(cls.puerto), PYTHONIOENCODING="utf-8")
|
|
cls.log = open(os.path.join(cls.tmp, "servidor.log"), "w", encoding="utf-8")
|
|
cls.proc = subprocess.Popen([sys.executable, "organizador.py", "--sin-navegador"], cwd=cls.app, env=env, stdout=cls.log, stderr=subprocess.STDOUT)
|
|
cls.base = f"http://127.0.0.1:{cls.puerto}"
|
|
for _ in range(60):
|
|
try:
|
|
urllib.request.urlopen(cls.base + "/api/estado", timeout=1)
|
|
break
|
|
except Exception: # noqa
|
|
time.sleep(0.25)
|
|
else:
|
|
raise RuntimeError("el servidor no arrancó: " + open(cls.log.name, encoding="utf-8", errors="replace").read()[-800:])
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.proc.kill()
|
|
cls.proc.wait()
|
|
cls.log.close()
|
|
shutil.rmtree(cls.tmp, ignore_errors=True)
|
|
|
|
def get(self, p):
|
|
return json.loads(urllib.request.urlopen(self.base + p, timeout=30).read())
|
|
|
|
def post(self, p, d):
|
|
rq = urllib.request.Request(self.base + p, json.dumps(d).encode(), {"Content-Type": "application/json", "Origin": self.base})
|
|
try:
|
|
r = urllib.request.urlopen(rq, timeout=60)
|
|
return r.status, json.loads(r.read())
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read())
|
|
|
|
def esperar_escaneo(self):
|
|
for _ in range(200):
|
|
r = next((x for x in self.get("/api/estado")["raices"] if x["ruta"].lower() == self.raiz.lower()), None)
|
|
if r and r["escaneo"] and not (r["progreso"] and r["progreso"]["estado"] == "corriendo"):
|
|
return r
|
|
time.sleep(0.1)
|
|
self.fail("el escaneo no terminó")
|
|
|
|
def _arbol(self):
|
|
out = []
|
|
for d, _, fs in os.walk(self.raiz):
|
|
for f in fs:
|
|
out.append(os.path.relpath(os.path.join(d, f), self.raiz).lower())
|
|
return sorted(out)
|
|
|
|
def test_una_sola_copia(self):
|
|
"""Abrir el programa otra vez sobre los mismos datos no levanta una segunda copia (se pisarían el
|
|
inventario, y la segunda daría por huérfanas las conversiones de la primera): se va, y dice por qué."""
|
|
with open(os.path.join(self.app, "datos", "sesion.json"), encoding="utf-8") as fh:
|
|
self.assertEqual(json.load(fh)["puerto"], self.puerto)
|
|
env = dict(os.environ, ORGANIZADOR_PUERTO=str(_puerto_libre()), PYTHONIOENCODING="utf-8")
|
|
p = subprocess.run([sys.executable, "organizador.py", "--sin-navegador"], cwd=self.app, env=env, capture_output=True,
|
|
text=True, encoding="utf-8", errors="replace", timeout=60)
|
|
self.assertEqual(p.returncode, 0)
|
|
self.assertIn("Ya hay un OrganiLAB abierto", p.stdout)
|
|
self.assertIn(str(self.puerto), p.stdout)
|
|
self.assertEqual(self.get("/api/estado")["version"], self.get("/api/actualizacion")["version"]) # la primera sigue atendiendo
|
|
|
|
def test_recorrido(self):
|
|
q = "?raiz=" + urllib.parse.quote(self.raiz)
|
|
# 0. una instalación nueva abre en blanco: ninguna carpeta puesta de fábrica
|
|
self.assertEqual(self.get("/api/estado")["raices"], [])
|
|
# peticiones que no son JSON: fuera
|
|
rq = urllib.request.Request(self.base + "/api/raices", b"{}", {"Content-Type": "text/plain"})
|
|
with self.assertRaises(urllib.error.HTTPError) as cm:
|
|
urllib.request.urlopen(rq)
|
|
self.assertEqual(cm.exception.code, 403)
|
|
# 1. registrar y escanear
|
|
self.assertEqual(self.post("/api/raices", {"raices": [{"ruta": self.raiz, "cliente": "Cliente Prueba", "tipo": "contenido"}]})[0], 200)
|
|
self.assertEqual(self.post("/api/escanear", {"raiz": self.raiz})[0], 200)
|
|
r = self.esperar_escaneo()
|
|
self.assertEqual(r["escaneo"]["n"], 6)
|
|
mapa = self.get("/api/mapa" + q)
|
|
self.assertTrue(mapa["escaneado"])
|
|
self.assertGreaterEqual(mapa["basura"]["n"], 1)
|
|
self.assertGreaterEqual(mapa["proxies"]["n"], 1)
|
|
# 2. plan: nada sale de la carpeta del cliente
|
|
plan = self.get("/api/plan" + q)
|
|
self.assertTrue(plan["filas"], "el plan no propuso nada")
|
|
for f in plan["filas"]:
|
|
self.assertTrue(os.path.normpath(f["destino"]).lower().startswith(os.path.normpath(self.raiz).lower()), f["destino"])
|
|
ids = [f["id"] for f in plan["filas"]]
|
|
# destino fuera de la carpeta: rechazado
|
|
c, d = self.post("/api/aplicar", {"raiz": self.raiz, "ids": ids, "destinos": {str(ids[0]): "..\\fuera"}})
|
|
self.assertEqual(c, 400)
|
|
self.assertIn("fuera", d["error"])
|
|
# 3. aplicar: el árbol cambia pero conserva los mismos archivos, y aparece _Para borrar
|
|
antes = self._arbol()
|
|
c, res = self.post("/api/aplicar", {"raiz": self.raiz, "ids": ids})
|
|
self.assertEqual(c, 200)
|
|
self.assertEqual(res["errores"], [])
|
|
self.assertGreater(res["movimientos"], 0)
|
|
despues = self._arbol()
|
|
self.assertNotEqual(antes, despues)
|
|
self.assertEqual(sorted(os.path.basename(p) for p in antes), sorted(os.path.basename(p) for p in despues))
|
|
self.assertTrue(any("_para borrar" in p for p in despues))
|
|
# 4. deshacer deja el árbol exacto
|
|
c, res = self.post("/api/deshacer", {"lote": res["lote"]})
|
|
self.assertEqual(c, 200)
|
|
self.assertEqual(antes, self._arbol())
|
|
# 4a. un mes escrito en minúsculas se normaliza de verdad (renombrado solo de mayúsculas), la
|
|
# carpeta nueva entra en el inventario sin volver a escanear, y deshacer lo devuelve
|
|
os.makedirs(os.path.join(self.raiz, "2026", "10- octubre"), exist_ok=True)
|
|
with open(os.path.join(self.raiz, "2026", "10- octubre", "Final.mp4"), "wb") as fh:
|
|
fh.write(os.urandom(4096))
|
|
self.post("/api/escanear", {"raiz": self.raiz})
|
|
self.esperar_escaneo()
|
|
pl = self.get("/api/plan" + q)
|
|
ren = [f for f in pl["filas"] if f["accion"] == "renombrar" and f["destino"].endswith("10- Octubre")]
|
|
self.assertTrue(ren, "el plan no propone normalizar «10- octubre»")
|
|
c, res = self.post("/api/aplicar", {"raiz": self.raiz, "ids": [f["id"] for f in ren]})
|
|
self.assertEqual(c, 200)
|
|
self.assertEqual(res["errores"], [])
|
|
self.assertIn("10- Octubre", os.listdir(os.path.join(self.raiz, "2026")))
|
|
ex = self.get("/api/estructura/existen?raiz=" + urllib.parse.quote(self.raiz)
|
|
+ "&ruta=" + urllib.parse.quote(json.dumps(["{anio}", "{mes}"])))
|
|
self.assertTrue(any(x.endswith("10- Octubre") for x in ex["carpetas"]), "la carpeta renombrada no está en el inventario")
|
|
c, _ = self.post("/api/deshacer", {"lote": res["lote"]})
|
|
self.assertIn("10- octubre", os.listdir(os.path.join(self.raiz, "2026")))
|
|
shutil.rmtree(os.path.join(self.raiz, "2026"), ignore_errors=True)
|
|
|
|
# 4a-bis. dos copias idénticas se ven como repetidas y una se aparta; dos con el mismo nombre y
|
|
# peso pero distinto contenido NO se tocan
|
|
os.makedirs(os.path.join(self.raiz, "Dup A"), exist_ok=True)
|
|
os.makedirs(os.path.join(self.raiz, "Dup B"), exist_ok=True)
|
|
cuerpo = os.urandom(4096) * 1500 # 6 MB: por encima del umbral
|
|
for d in ("Dup A", "Dup B"):
|
|
with open(os.path.join(self.raiz, d, "igual.mp4"), "wb") as fh:
|
|
fh.write(cuerpo)
|
|
with open(os.path.join(self.raiz, "Dup A", "parecido.mp4"), "wb") as fh:
|
|
fh.write(os.urandom(4096) * 1500)
|
|
with open(os.path.join(self.raiz, "Dup B", "parecido.mp4"), "wb") as fh:
|
|
fh.write(os.urandom(4096) * 1500)
|
|
self.post("/api/escanear", {"raiz": self.raiz})
|
|
self.esperar_escaneo()
|
|
dup = self.get("/api/duplicados" + q)
|
|
self.assertEqual(len(dup["grupos"]), 2)
|
|
c, ap = self.post("/api/duplicados/apartar", {"raiz": self.raiz})
|
|
self.assertEqual(ap["movimientos"], 1) # solo la idéntica
|
|
self.assertEqual(ap["distintos"], 1) # la parecida se deja, y se avisa
|
|
shutil.rmtree(os.path.join(self.raiz, "Dup A"), ignore_errors=True)
|
|
shutil.rmtree(os.path.join(self.raiz, "Dup B"), ignore_errors=True)
|
|
self.post("/api/escanear", {"raiz": self.raiz})
|
|
self.esperar_escaneo()
|
|
|
|
# 4b. el mapa dice dónde va a acabar cada cosa, con las plantillas de verdad
|
|
est = self.get("/api/estructura" + q)
|
|
proy = next(c for c in est["cubetas"] if c["clave"] == "PROYECTOS")
|
|
self.assertIn("{app}", proy["plantilla"])
|
|
self.assertTrue(proy["guarda"])
|
|
nombres = []
|
|
|
|
def bajar(n):
|
|
nombres.append(n["nombre"])
|
|
for h in n["hijos"]:
|
|
bajar(h)
|
|
bajar(est["arbol"])
|
|
self.assertIn("00_Grabacion", nombres)
|
|
self.assertIn("Premiere", nombres)
|
|
# guardar UNA plantilla no puede llevarse por delante las demás
|
|
antes_pl = len([c for c in est["cubetas"] if c["plantilla"]])
|
|
self.assertEqual(self.post("/api/plantillas", {"plantillas": {"contenido": {"MUSICA": "_Recursos/Musica/{sub}"}}})[0], 200)
|
|
self.assertEqual(len([c for c in self.get("/api/estructura" + q)["cubetas"] if c["plantilla"]]), antes_pl)
|
|
|
|
# 4b-bis. el mapa al revés: dónde iría un archivo concreto
|
|
def donde(nombre):
|
|
return self.get("/api/estructura/archivo?tipo=contenido&nombre=" + urllib.parse.quote(nombre))
|
|
d = donde("A006_12171629_C001.braw")
|
|
self.assertEqual(d["cubeta"], "GRABACION")
|
|
self.assertIn("URSA_BRAW", d["destino"])
|
|
self.assertIn("12- Diciembre", d["destino"]) # la fecha sale del nombre de la cámara
|
|
self.assertEqual(donde("Proyecto.aep")["destino"], os.path.join("01_Proyectos", "After Effects"))
|
|
self.assertEqual(donde("cancion.wav")["cubeta"], "MUSICA")
|
|
self.assertEqual(d["dia"], "17- Miércoles") # el 17-dic-2025 fue miércoles
|
|
self.assertEqual(donde("cancion.wav")["dia"], "") # el día solo aplica a grabaciones
|
|
self.assertTrue(donde("")["error"])
|
|
|
|
# 4b-ter. renombrar una carpeta del mapa cambia TODAS las plantillas que pasan por ella,
|
|
# y las carpetas que ya existen en el disco si se pide
|
|
os.makedirs(os.path.join(self.raiz, "2026", "03- Marzo", "00_Grabacion"), exist_ok=True)
|
|
self.post("/api/escanear", {"raiz": self.raiz})
|
|
self.esperar_escaneo()
|
|
crudo = ["{anio}", "{mes}", "00_Grabacion"]
|
|
ex = self.get("/api/estructura/existen?raiz=" + urllib.parse.quote(self.raiz)
|
|
+ "&ruta=" + urllib.parse.quote(json.dumps(crudo)))
|
|
self.assertEqual(ex["n"], 1)
|
|
# «{anio}» solo vale por algo que parezca un año
|
|
self.assertEqual(self.get("/api/estructura/existen?raiz=" + urllib.parse.quote(self.raiz)
|
|
+ "&ruta=" + urllib.parse.quote(json.dumps(["{anio}"])))["n"], 1)
|
|
c, r = self.post("/api/estructura/renombrar", {"tipo": "contenido", "ruta": crudo,
|
|
"nombre": "Grabacion", "raiz": self.raiz, "en_disco": True})
|
|
self.assertEqual(c, 200)
|
|
self.assertTrue(r.get("ok"), r)
|
|
self.assertEqual(r["movidas"], 1)
|
|
self.assertIn("CELULAR", r["cubetas"]) # la de celular cuelga de la misma carpeta
|
|
self.assertTrue(os.path.isdir(os.path.join(self.raiz, "2026", "03- Marzo", "Grabacion")))
|
|
for malo in ("", "con/barra", "{anio}"):
|
|
self.assertTrue(self.post("/api/estructura/renombrar",
|
|
{"tipo": "contenido", "ruta": crudo, "nombre": malo})[1].get("error"))
|
|
shutil.rmtree(os.path.join(self.raiz, "2026"), ignore_errors=True)
|
|
|
|
# 4b-quáter. colores de carpeta en Windows: se ponen y se quitan sin tocar ningún archivo
|
|
self.post("/api/escanear", {"raiz": self.raiz})
|
|
self.esperar_escaneo()
|
|
arbol_color = self._arbol()
|
|
c, col = self.post("/api/colores/pintar", {"raiz": self.raiz})
|
|
self.assertEqual(c, 200)
|
|
if not col.get("error"): # algunos discos no lo permiten: es válido
|
|
self.assertGreater(col["pintadas"], 0)
|
|
self.assertTrue(os.path.isdir(os.path.join(self.raiz, R.CARPETA_APP, "iconos")))
|
|
self.assertEqual(arbol_color, self._arbol()) # ni un archivo del cliente tocado
|
|
c, q = self.post("/api/colores/quitar", {"raiz": self.raiz})
|
|
self.assertEqual(q["quitadas"], col["pintadas"])
|
|
self.assertFalse(os.path.exists(os.path.join(self.raiz, R.CARPETA_APP)))
|
|
self.assertEqual(arbol_color, self._arbol())
|
|
|
|
# 4c. carpetas vacías: se encuentran, se borran sin tocar un archivo y deshacer las devuelve
|
|
for rel in ("Huecos/uno", "Huecos/dos", "Sola"):
|
|
os.makedirs(os.path.join(self.raiz, rel.replace("/", os.sep)), exist_ok=True)
|
|
self.post("/api/escanear", {"raiz": self.raiz})
|
|
self.esperar_escaneo()
|
|
v = self.get("/api/vacias" + q)
|
|
rels = {c["rel"] for c in v["carpetas"]}
|
|
self.assertIn("Huecos", rels) # el tope, no cada hija por separado
|
|
self.assertIn("Sola", rels)
|
|
self.assertNotIn(os.path.join("Huecos", "uno"), rels)
|
|
arbol_antes = self._arbol()
|
|
c, res = self.post("/api/vacias/borrar", {"raiz": self.raiz, "rutas": [x["ruta"] for x in v["carpetas"]]})
|
|
self.assertEqual(c, 200)
|
|
self.assertEqual(res["errores"], [])
|
|
self.assertEqual(res["borradas"], 4) # Huecos con sus dos hijas, y Sola
|
|
self.assertFalse(os.path.isdir(os.path.join(self.raiz, "Sola")))
|
|
self.assertEqual(arbol_antes, self._arbol()) # ni un archivo tocado
|
|
c, _ = self.post("/api/deshacer", {"lote": res["lote"]})
|
|
self.assertEqual(c, 200)
|
|
self.assertTrue(os.path.isdir(os.path.join(self.raiz, "Huecos", "uno")))
|
|
self.post("/api/vacias/borrar", {"raiz": self.raiz, "rutas": [x["ruta"] for x in v["carpetas"]]})
|
|
|
|
# 5. desconectar con datos; la segunda vez ya no existe
|
|
c, d = self.post("/api/raiz/quitar", {"raiz": self.raiz, "datos": True})
|
|
self.assertEqual(c, 200)
|
|
self.assertEqual(self.get("/api/estado")["raices"], [])
|
|
self.assertEqual(self.post("/api/raiz/quitar", {"raiz": self.raiz})[0], 409)
|
|
# 6. actualizaciones automáticas: apagadas por defecto; encenderlas deja constancia de que fue a mano
|
|
act = self.get("/api/estado")["config"].get("actualizaciones") or {}
|
|
self.assertIsNot(act.get("auto"), True)
|
|
c, r = self.post("/api/config", {"actualizaciones": {"auto": True}})
|
|
self.assertTrue(r["config"]["actualizaciones"]["auto"] and r["config"]["actualizaciones"]["elegido"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|