9d4d1e9cc2
El año y el mes salían de la fecha de cada archivo desde 2.2.0, pero dentro
del mes iba todo junto: en Camilo, 13 días de rodaje y 907 archivos en la
misma carpeta.
- Nueva variable {dia} («15- Viernes», como los meses) en las plantillas de
grabación de los cuatro tipos de trabajo, con migración de las guardadas.
- Plan._dias_de_rodaje() cuenta los días distintos por mes y dia_de()
devuelve "" cuando el mes tuvo un solo rodaje: la carpeta del día solo
aparece donde hace falta. David Reyes la necesita en 1 de 8 meses,
Danney en 36 de 57, Camilo en 12 de 26. 186 grupos, cero colisiones.
- Un archivo de cámara suelto también se archiva por su propia fecha:
plan_archivo hacía «if mes: fm = mes» y se saltaba la regla de 2.2.0.
- «¿Dónde va a parar este archivo?» dice también el día.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
326 líneas
16 KiB
Python
326 líneas
16 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())
|
|
# 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)
|