# -*- 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) # 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)