Labstream/labstream-organizador
1
0
Fork 0
Código Incidencias Pull Requests Paquetes Proyectos Lanzamientos Wiki Actividad
Files
477857db45182327f1ccccc38a86ae7c2c5261f1
labstream-organizador/probar.py
Labstream 477857db45 3.3.0: lo que entra nuevo ya entra ordenado 
Las cuatro funciones que eligió Munka tras la 3.2.0:

- Nueva pieza: crea «NN- Título» en su mes con el número que toca y sus subcarpetas; si la
  pieza ya existe, lo dice. Se deshace mientras siga vacía.
- Bandeja de entrada: lo que se deje en _Entrada se reparte por mes, día de rodaje y cámara.
  «Traer material» copia una tarjeta comprobando cada archivo (se vuelve a leer y se compara
  la huella), conserva la fecha del rodaje y no toca el origen.
- Versiones del mismo clip: dos clips de cámara que se llaman igual, duran lo mismo y tienen
  la misma forma son el mismo clip; se queda el de más imagen y el otro va a Duplicados.
- Orden automático, apagado de fábrica: una tarea del Programador de tareas de Windows lanza
  «automatico», que lee todas las carpetas y hace solo lo seguro. No toca una carpeta que no
  se pueda leer entera ni una cuyo plan mueva más de 3.000 archivos, y si el programa ya está
  abierto se lo pide a él. Desinstalar quita la tarea.

De paso: una ruta que empieza por guion bajo ya no pierde el guion al ir a la papelera.

Pruebas: prueba_orden.py 24, probar.py 19, prueba_funcional.py 29, prueba_ventana.py 18
desde el código y 17 con el paquete.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-04 17:05:57 -05:00

644 líneas
36 KiB
Python
Original Blame Histórico

# -*- 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", "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", "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)
Referencia en una nueva incidencia Ver la culpa de Git Copiar Permalink