1ae7a1bb38
- Cada carpeta lleva el color de lo que guarda (R.COLOR_CUBETA). Lo que
cuelga de una carpeta con color hereda el suyo; una que mezcla tipos se
queda neutra.
- «¿Dónde va a parar este archivo?»: donde_iria() resuelve cualquier nombre
con el mismo clasificador del escaneo y el árbol enciende el destino.
Debajo, nombres de archivo de verdad de la carpeta conectada.
- Renombrar una carpeta del árbol cambia TODAS las plantillas que pasan por
ella (00_Grabacion arrastra a CELULAR) y, si se pide, las carpetas que ya
existen en el disco; queda en la bitácora y se deshace.
- {anio} y {mes} dejan de encajar con cualquier carpeta: solo con algo que
parezca un año o un mes.
- Colores en el Explorador: .ico generados sin dependencias y asignados con
SHGetSetFolderCustomSettings, la misma API de «Cambiar icono». Escribir el
desktop.ini a mano NO basta, y SHGetFileInfo no sirve para comprobarlo:
no informa de la personalización ni cuando la puso el propio Explorador.
Ruta de icono relativa, así que los colores viajan con el disco. Si el
disco no deja marcar la carpeta, se dice y no se deja nada escrito.
- .organilab no se escanea y los desktop.ini de la app no son basura.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
317 líneas
16 KiB
Python
317 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_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.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)
|