Files
labstream-organizador/probar.py
Labstream 1151320bf2 2.6.0: auditoría, 100 % local y 26 funciones verificadas
Sale a la red solo lo que se pide: la tipografía viene con el programa
(static/manrope.woff2) y el comprobador de versiones tiene interruptor en
Ajustes. Nueva prueba funcional (prueba_funcional.py) que levanta una copia
sobre un cliente de ensayo y ejercita las 26 funciones por su API, con una
conversión real: 26 de 26.

Fallos que sacó y quedan arreglados:
- Un renombrado solo de mayúsculas («10- octubre» → «10- Octubre») se
  descartaba en Plan.agregar y en mover; deshacer decía «sitio ocupado».
- Las carpetas que crea el plan no entraban en el inventario hasta el
  siguiente escaneo (_recontar_carpetas las da de alta y recuenta).
- Una fusión de carpetas cortaba en el primer archivo que tropezaba.
- «Viejo» era distinto en convertidor, stock y respaldo: ahora es_viejo.
- /api/actualizacion informaba versión vacía hasta la primera consulta.
- Filas de sueltos por carpeta de origen; mes del stock el más repetido;
  fuera tres rutas sin uso.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-24 20:57:41 -05:00

371 líneas
19 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 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"):
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_recorrido(self):
q = "?raiz=" + urllib.parse.quote(self.raiz)
# 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)
if __name__ == "__main__":
unittest.main(verbosity=2)