# -*- coding: utf-8 -*- """Prueba funcional completa: levanta una copia del programa sobre un cliente de ensayo con de todo (cámara, exportados, proyectos, música, stock de Envato, basura, proxies, repetidos, huecos, un mes mal escrito y un clip de video de verdad) y ejercita TODAS las funciones por su API, en el mismo orden en que las usaría una persona. Imprime una tabla: qué se probó, qué salió y en cuánto tiempo.""" import os import sys import json import time import shutil import socket import tempfile import subprocess import urllib.parse import urllib.error import urllib.request AQUI = os.path.dirname(os.path.abspath(__file__)) TMP = tempfile.mkdtemp(prefix='organilab-smoke-') APP = os.path.join(TMP, 'app') os.makedirs(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), APP) shutil.copytree(os.path.join(AQUI, 'static'), os.path.join(APP, 'static')) RAIZ = os.path.join(TMP, 'Cliente Smoke') RESP = os.path.join(TMP, 'Respaldo') FILAS = [] def prueba(nombre, fn): t = time.time() try: det = fn() FILAS.append((nombre, 'OK', '%.2f s' % (time.time() - t), det or '')) except Exception as ex: # noqa FILAS.append((nombre, 'FALLA', '%.2f s' % (time.time() - t), str(ex)[:160])) def mk(rel, y, m, d, tam=4096): p = os.path.join(RAIZ, rel.replace('/', os.sep)) os.makedirs(os.path.dirname(p), exist_ok=True) with open(p, 'wb') as fh: fh.write(os.urandom(4096) * max(1, tam // 4096)) t = time.mktime(time.struct_time((y, m, d, 12, 0, 0, 0, 0, -1))) os.utime(p, (t, t)) return p # ---------------------------------------------------------------- el cliente de ensayo mk('2026/03- Marzo/Grabacion/A001_03051200_C001.braw', 2026, 3, 5, 3_000_000) mk('2026/03- Marzo/Grabacion/A001_03051200_C002.braw', 2026, 3, 5, 3_000_000) mk('2026/03- Marzo/Grabacion/C0010.MP4', 2026, 3, 12, 2_000_000) # sony, otro día mk('2026/03- Marzo/Grabacion/C0010M01.XML', 2026, 3, 12, 500) mk('2026/03- Marzo/Grabacion/Proxy/C0010.MOV', 2026, 3, 12, 100_000) # proxy mk('2026/03- Marzo/Grabacion/Thumbs.db', 2026, 3, 12, 100) # basura mk('2026/03- Marzo/01- Botox/Reel botox.mp4', 2026, 3, 20, 6_000_000) mk('2026/03- Marzo/01- Botox/Reel botox.prproj', 2026, 3, 20, 50_000) mk('2026/03- Marzo/01- Botox/Visual/Elements/clinic-2025-01-22-22-34-00-utc.mp4', 2026, 3, 20, 900_000) # stock viejo mk('2026/10- octubre/Exportados/Final.mp4', 2026, 10, 2, 700_000) # mes en minúsculas mk('Musica/tema.wav', 2026, 1, 1, 300_000) mk('Sueltos/Portada.jpg', 2026, 2, 2, 60_000) mk('Sueltos/Pieza.aep', 2026, 2, 2, 40_000) shutil.copy2(os.path.join(RAIZ, '2026', '03- Marzo', '01- Botox', 'Reel botox.mp4'), os.path.join(RAIZ, 'Sueltos', 'Reel botox.mp4')) # repetido de verdad: mismo contenido mk('Raro/algo.xyz', 2026, 2, 2, 10) os.makedirs(os.path.join(RAIZ, 'Vacia', 'dentro'), exist_ok=True) # un video de verdad para el convertidor (6 s, 1280x720, H.264) FF = shutil.which('ffmpeg') CLIP = os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'C0020.MP4') if FF: subprocess.run([FF, '-hide_banner', '-loglevel', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=6:size=1280x720:rate=25', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=6', '-c:v', 'libx264', '-preset', 'ultrafast', '-crf', '18', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', CLIP], check=True) t = time.mktime(time.struct_time((2025, 6, 10, 12, 0, 0, 0, 0, -1))) os.utime(CLIP, (t, t)) # ---------------------------------------------------------------- el servidor with socket.socket() as s_: s_.bind(('127.0.0.1', 0)) PUERTO = s_.getsockname()[1] env = dict(os.environ, ORGANIZADOR_PUERTO=str(PUERTO), PYTHONIOENCODING='utf-8') log = open(os.path.join(TMP, 'servidor.log'), 'w', encoding='utf-8') proc = subprocess.Popen([sys.executable, 'organizador.py', '--sin-navegador'], cwd=APP, env=env, stdout=log, stderr=subprocess.STDOUT) BASE = f'http://127.0.0.1:{PUERTO}' def get(p): return json.loads(urllib.request.urlopen(BASE + p, timeout=60).read()) def post(p, d): rq = urllib.request.Request(BASE + p, json.dumps(d).encode(), {'Content-Type': 'application/json', 'Origin': BASE}) try: r = urllib.request.urlopen(rq, timeout=120) return r.status, json.loads(r.read()) except urllib.error.HTTPError as e: return e.code, json.loads(e.read()) def q(**kw): return '?' + urllib.parse.urlencode(kw) for _ in range(80): try: urllib.request.urlopen(BASE + '/api/estado', timeout=1) break except Exception: # noqa time.sleep(0.25) else: raise SystemExit('el servidor no arrancó') def esperar(cond, seg=90): fin = time.time() + seg while time.time() < fin: if cond(): return True time.sleep(0.3) return False def raiz_estado(): return next((x for x in get('/api/estado')['raices'] if x['ruta'].lower() == RAIZ.lower()), None) def arbol(): return sorted(os.path.relpath(os.path.join(b, f), RAIZ).lower() for b, _, fs in os.walk(RAIZ) for f in fs if '.organilab' not in b.lower()) # ---------------------------------------------------------------- las pruebas, en orden de uso def p_estatico(): html = urllib.request.urlopen(BASE + '/', timeout=10).read().decode('utf-8') assert 'googleapis' not in html, 'la interfaz sigue pidiendo la fuente a Google' r = urllib.request.urlopen(BASE + '/static/manrope.woff2', timeout=10) assert r.headers.get('Content-Type') == 'font/woff2', r.headers.get('Content-Type') assert r.read(4) == b'wOF2' return 'sin nada externo; fuente local servida como font/woff2' def p_registrar(): c, r = post('/api/raices', {'raices': [{'ruta': RAIZ, 'cliente': 'Cliente Smoke', 'tipo': 'contenido'}]}) assert c == 200, r # anidada: se rechaza c2, r2 = post('/api/raices', {'raices': [{'ruta': RAIZ, 'cliente': 'A', 'tipo': 'contenido'}, {'ruta': os.path.join(RAIZ, 'Sueltos'), 'cliente': 'B', 'tipo': 'contenido'}]}) assert c2 != 200 or r2.get('error'), 'aceptó una carpeta dentro de otra' return 'registra, y rechaza una carpeta anidada' def p_escanear(): c, r = post('/api/escanear', {'raiz': RAIZ}) assert c == 200 and r.get('lanzado'), r assert esperar(lambda: (raiz_estado() or {}).get('escaneo') and not ((raiz_estado() or {}).get('progreso') or {}).get('estado') == 'corriendo') e = raiz_estado()['escaneo'] esperados = len(arbol()) assert e['n'] == esperados, 'inventarió %d de %d' % (e['n'], esperados) return '%d archivos inventariados, los mismos que hay en el disco' % e['n'] def p_carpeta(): e = get('/api/carpeta/estado' + q(raiz=RAIZ)) assert e.get('raiz') and 'escaneo' in e, e a = get('/api/carpeta/acciones' + q(raiz=RAIZ)) claves = [x['clave'] for x in a['acciones']] for k in ('limpiar', 'nombrar', 'fechas', 'resto', 'vacias', 'stock', 'repetidos', 'respaldar'): assert k in claves, 'falta la acción ' + k toca = [x['clave'] for x in a['acciones'] if x['estado'] == 'toca'] assert 'limpiar' in toca and 'vacias' in toca and 'repetidos' in toca and 'stock' in toca, toca return 'lista de acciones completa; pendientes: ' + ', '.join(toca) def p_que_hay(): m = get('/api/mapa' + q(raiz=RAIZ)) assert m['basura']['n'] >= 1 and m['proxies']['n'] >= 1, (m['basura'], m['proxies']) assert m.get('arbol'), 'sin árbol de peso' d = get('/api/discos') assert isinstance(d, (list, dict)) return 'basura %d · proxies %d · %d ramas' % (m['basura']['n'], m['proxies']['n'], len(m['arbol'])) def p_nombrar(): v = get('/api/vocabulario' + q(raiz=RAIZ)) toks = {t['token']: t for t in v['tokens']} assert 'raro' in toks and not toks['raro'].get('cubeta'), 'debería salir «Raro» como no reconocida' c, r = post('/api/diccionario', {'token': 'raro', 'cubeta': 'DOCUMENTOS'}) assert c == 200, r v2 = get('/api/vocabulario' + q(raiz=RAIZ)) assert {t['token']: t for t in v2['tokens']}['raro'].get('cubeta') == 'DOCUMENTOS' c, r = post('/api/diccionario', {'token': 'raro', 'cubeta': None}) # y se olvida # las carpetas de la propia app no se enseñan: «basura» es su _Para borrar\basura, no una del cliente c, r = post('/api/diccionario', {'token': 'Basura', 'cubeta': 'GRABACION'}) assert c == 400 and 'propia app' in r.get('error', ''), (c, r) return 'reconoce, aprende y olvida un nombre de carpeta; no deja enseñar las de la app' PLAN = {} def p_plan(): pl = get('/api/plan' + q(raiz=RAIZ)) PLAN.update(pl) filas = pl['filas'] assert filas, 'plan vacío' cub = {f['cubeta'] for f in filas} for c in ('GRABACION', 'BASURA', 'PROXY'): assert c in cub, 'el plan no propone nada de ' + c mes = [f for f in filas if f['accion'] == 'renombrar' and '10- Octubre' in f['destino']] assert mes, 'no propone normalizar «10- octubre»' dias = {f['destino_rel'].split(os.sep)[3] for f in filas if f['cubeta'] == 'GRABACION' and f['destino_rel'].count(os.sep) >= 3} assert any(d.startswith(('05-', '12-')) for d in dias), 'marzo tuvo dos rodajes y no sale carpeta de día: %s' % dias return '%d filas · %d marcadas · reparte marzo por día de rodaje' % (len(filas), sum(1 for f in filas if f['marcado'])) LOTES = {} def p_aplicar(): antes = arbol() c, r = post('/api/aplicar', {'raiz': RAIZ, 'ids': [f['id'] for f in PLAN['filas'] if f['marcado']]}) assert c == 200, r assert not r['errores'], r['errores'] assert r['movimientos'] > 0, 'no movió nada' LOTES['orden'] = r['lote'] despues = arbol() assert sorted(os.path.basename(x) for x in antes) == sorted(os.path.basename(x) for x in despues), 'se perdió algún archivo' assert '10- Octubre' in os.listdir(os.path.join(RAIZ, '2026')), 'el mes sigue en minúsculas: %s' % os.listdir(os.path.join(RAIZ, '2026')) assert not any('thumbs.db' in x and '_para borrar' not in x for x in despues), 'la basura sigue fuera' return '%d movimientos, sin errores, «10- octubre» pasó a «10- Octubre»' % r['movimientos'] def p_deshacer(): c, r = post('/api/deshacer', {'lote': LOTES['orden']}) assert c == 200 and not r['errores'], r assert '10- octubre' in os.listdir(os.path.join(RAIZ, '2026')), 'deshacer no devolvió las minúsculas: %s' % os.listdir(os.path.join(RAIZ, '2026')) assert os.path.isfile(os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'Thumbs.db')), 'la basura no volvió' return '%d movimientos deshechos' % r['deshechos'] def p_reaplicar(): pl = get('/api/plan' + q(raiz=RAIZ)) c, r = post('/api/aplicar', {'raiz': RAIZ, 'ids': [f['id'] for f in pl['filas'] if f['marcado']]}) assert c == 200 and not r['errores'], r LOTES['orden2'] = r['lote'] return 'vuelve a aplicar sin errores' def p_historial(): b = get('/api/bitacora' + q(raiz=RAIZ)) lotes = b.get('lotes') or b assert any(LOTES['orden2'] in json.dumps(l) for l in lotes), 'el lote aplicado no sale en el historial' p = get('/api/papelera' + q(raiz=RAIZ)) assert any(x['nombre'] == 'basura' and x['n'] >= 1 for x in p['subcarpetas']), p return 'historial y _Para borrar reflejan lo hecho' def p_repetidos(): d = get('/api/duplicados' + q(raiz=RAIZ)) assert d['grupos'], 'no ve el Reel botox repetido' c, r = post('/api/duplicados/apartar', {'raiz': RAIZ}) assert c == 200 and r['movimientos'] >= 1, r return '%d grupo(s), %d copia(s) apartada(s) a Duplicados' % (len(d['grupos']), r['movimientos']) def p_stock(): s = get('/api/stock' + q(raiz=RAIZ, meses=3, tipos='Video')) assert s['grupos'], 'no ve el stock de Envato viejo' rutas = [x for g in s['grupos'] for x in g['rutas']] c, r = post('/api/stock/papelera', {'raiz': RAIZ, 'rutas': rutas}) assert c == 200 and r['movimientos'] == len(rutas), r return '%d archivo(s) de stock apartados' % r['movimientos'] def p_vacias(): v = get('/api/vacias' + q(raiz=RAIZ)) assert any(x['rel'] == 'Vacia' for x in v['carpetas']), v c, r = post('/api/vacias/borrar', {'raiz': RAIZ, 'rutas': [x['ruta'] for x in v['carpetas']]}) assert c == 200 and r['borradas'] >= 2, r c, d = post('/api/deshacer', {'lote': r['lote']}) assert os.path.isdir(os.path.join(RAIZ, 'Vacia', 'dentro')), 'deshacer no recreó la carpeta' post('/api/vacias/borrar', {'raiz': RAIZ, 'rutas': [x['ruta'] for x in v['carpetas']]}) return '%d huecos borrados, deshechos y vueltos a borrar' % r['borradas'] def p_mapa(): e = get('/api/estructura' + q(raiz=RAIZ)) assert e['cubetas'] and e['arbol']['hijos'], 'mapa vacío' a = get('/api/estructura/archivo' + q(nombre='Proyecto.aep', tipo='contenido')) assert a['destino'].endswith('After Effects'), a ex = get('/api/estructura/existen' + q(raiz=RAIZ, ruta=json.dumps(['{anio}', '{mes}', '00_Grabacion']))) assert ex['n'] >= 1, ex c, r = post('/api/estructura/renombrar', {'tipo': 'contenido', 'ruta': ['{anio}', '{mes}', '00_Grabacion'], 'nombre': 'Grabacion', 'raiz': RAIZ, 'en_disco': True}) assert c == 200 and r.get('ok') and r['movidas'] >= 1, r c, r2 = post('/api/estructura/renombrar', {'tipo': 'contenido', 'ruta': ['{anio}', '{mes}', 'Grabacion'], 'nombre': '00_Grabacion', 'raiz': RAIZ, 'en_disco': True}) assert r2.get('ok'), r2 c, r3 = post('/api/plantillas', {'plantillas': {'contenido': {'MUSICA': '_Recursos/Musica/{sub}'}}}) assert c == 200 return 'árbol, dónde iría, existen, renombrar ida y vuelta, guardar plantilla' def p_colores(): c, r = post('/api/colores/pintar', {'raiz': RAIZ}) assert c == 200, r if r.get('error'): return 'este disco no lo permite (' + r['error'][:50] + ')' assert r['pintadas'] >= 3, r c, qq = post('/api/colores/quitar', {'raiz': RAIZ}) assert qq['quitadas'] == r['pintadas'], (r, qq) return '%d carpetas pintadas y despintadas' % r['pintadas'] IDS = {} def p_motores(): m = get('/api/optimizar/motores') assert m.get('ffmpeg'), 'sin ffmpeg' disp = [k for k, v in m['formatos'].items() if v] assert 'h265_10bit' in disp or 'h264' in disp, disp return 'formatos disponibles: ' + ', '.join(disp) + ' · GPU: ' + (m['gpu'].get('nombre') or 'no') def p_analizar(): post('/api/config', {'optimizar': {'meses_minimos': 0, 'min_mb': 0}}) c, r = post('/api/optimizar/analizar', {'raiz': RAIZ, 'min_mb': 0}) assert c == 200, r assert esperar(lambda: ((raiz_estado() or {}).get('analisis') or {}).get('estado') not in (None, 'corriendo'), 120) cand = get('/api/optimizar/candidatos' + q(raiz=RAIZ, meses=0, min_mb=0, formato='h265_10bit'))['candidatos'] clip = [x for x in cand if x['ruta'].lower().endswith('c0020.mp4')] assert clip and not clip[0]['excluido'], (clip or cand)[:1] IDS['ruta'] = clip[0]['ruta'] return 'leyó %d clip(s) · C0020.MP4 %s %s' % (len(cand), clip[0]['codec'], clip[0]['res']) def p_convertir(): m = get('/api/optimizar/motores') fmt = 'h265_10bit' if m['formatos'].get('h265_10bit') else 'h264' c, r = post('/api/optimizar/encolar', {'raiz': RAIZ, 'rutas': [IDS['ruta']], 'perfil': 'equilibrado', 'formato': fmt, 'chroma': '420', 'color': 'rec709', 'iniciar': True}) assert c == 200 and r['ids'], r IDS['trabajo'] = r['ids'][0] assert esperar(lambda: not get('/api/estado')['cola']['activa'], 240), 'la cola no terminó' cola = get('/api/optimizar/cola' + q(raiz=RAIZ)) t = next(x for x in cola['trabajos'] if x['id'] == IDS['trabajo']) assert t['estado'] in ('listo', 'verificado'), (t['estado'], t.get('log', '')[-200:]) c, v = post('/api/optimizar/verificar', {'ids': [IDS['trabajo']]}) assert v['resultados'][0]['ok'], v c, rr = post('/api/optimizar/reemplazar', {'ids': [IDS['trabajo']]}) assert rr['resultados'][0]['ok'], rr assert os.path.isfile(t['salida']) and not os.path.isfile(IDS['ruta']) return '%s: convertido (%s), verificado y reemplazado; original apartado' % (fmt, t['motor']) def p_cola_control(): c, r = post('/api/optimizar/control', {'accion': 'quitar', 'ids': [IDS['trabajo']]}) assert c == 200 return 'control de cola responde' def p_respaldo(): pl = get('/api/respaldo/plan' + q(raiz=RAIZ, destino=RESP, por_tipo=1, sin_stock=1, incluir_raw=0, meses=0)) assert pl.get('n', 0) > 0, pl c, r = post('/api/respaldo/copiar', {'raiz': RAIZ}) assert c == 200, r assert esperar(lambda: get('/api/respaldo/estado')['estado'] != 'corriendo', 120) e = get('/api/respaldo/estado') assert e['hechos'] == pl['n'], (e['hechos'], pl['n']) assert os.path.isfile(os.path.join(RESP, 'Cliente Smoke', 'INDICE.csv')), 'falta INDICE.csv' return '%d archivos copiados a %s · INDICE.csv' % (e['hechos'], os.path.basename(RESP)) def p_asistente(): pz = get('/api/asistente/pasos' + q(raiz=RAIZ)) assert 'pasos' in pz and 'diagnostico' in pz, list(pz) c, r = post('/api/acciones/hacer', {'raiz': RAIZ}) assert c == 200, r assert esperar(lambda: not get('/api/estado')['asistente'].get('activo'), 120) a = get('/api/estado')['asistente'] assert not a.get('error'), a.get('error') return 'pasos + «hacer las N de una vez» terminan sin error' def p_soltar_y_buscar(): s = get('/api/soltado' + q(nombre='nada')) assert s['ruta'] is None b = get('/api/buscar-carpeta' + q(nombre='Cliente Smoke')) assert isinstance(b['coincidencias'], list) return 'puente de arrastre responde; búsqueda de carpeta responde' def p_actualizacion_offline(): # con una URL imposible el programa no puede quedarse colgado ni romper nada a = get('/api/actualizacion') assert a['version'] c, r = post('/api/config', {'actualizaciones': {'auto': False}}) assert c == 200 and r['config']['actualizaciones']['auto'] is False, r return 'estado legible; el interruptor de arranque se guarda' def p_clientes_config(): e = get('/api/estado') cl = dict(e['config']['clientes']) cl['Cliente Smoke'] = {'tipo': 'contenido', 'alias': ['smoke']} c, r = post('/api/clientes', {'clientes': cl}) assert c == 200 c, r = post('/api/regla', {'cliente': 'Cliente Smoke', 'origen': 'Sueltos', 'destino': '_Revisar/Sueltos', 'cubeta': 'REVISAR'}) assert c == 200 return 'clientes y reglas se guardan' def p_quitar(): c, r = post('/api/raiz/quitar', {'raiz': RAIZ, 'datos': True}) assert c == 200, r assert get('/api/estado')['raices'] == [] return 'desconectada con sus datos' for nombre, fn in (('Interfaz sin nada externo', p_estatico), ('Registrar carpeta', p_registrar), ('Escanear', p_escanear), ('Estado y lista de acciones', p_carpeta), ('Qué hay (mapa de peso, discos)', p_que_hay), ('Por nombrar (vocabulario, diccionario)', p_nombrar), ('Calcular el plan', p_plan), ('Aplicar el plan', p_aplicar), ('Deshacer', p_deshacer), ('Volver a aplicar', p_reaplicar), ('Historial y papelera', p_historial), ('Repetidos', p_repetidos), ('Stock viejo', p_stock), ('Carpetas vacías', p_vacias), ('Mapa interactivo', p_mapa), ('Colores en Windows', p_colores), ('Motores de conversión', p_motores), ('Leer clips (ffprobe)', p_analizar), ('Convertir, verificar, reemplazar', p_convertir), ('Control de la cola', p_cola_control), ('Respaldar', p_respaldo), ('Asistente / hacer de una vez', p_asistente), ('Arrastrar y buscar carpeta', p_soltar_y_buscar), ('Actualizaciones', p_actualizacion_offline), ('Clientes, reglas', p_clientes_config), ('Quitar de la lista', p_quitar)): prueba(nombre, fn) proc.kill() proc.wait() log.close() print('%-40s %-6s %-8s %s' % ('FUNCIÓN', 'ESTADO', 'TIEMPO', 'DETALLE')) print('-' * 120) for n, e, t, d in FILAS: print('%-40s %-6s %-8s %s' % (n, e, t, d)) malas = [f for f in FILAS if f[1] != 'OK'] print('-' * 120) print('%d de %d funciones OK' % (len(FILAS) - len(malas), len(FILAS))) if malas: print() print('== últimas líneas del servidor ==') print(open(os.path.join(TMP, 'servidor.log'), encoding='utf-8', errors='replace').read()[-1500:]) shutil.rmtree(TMP, ignore_errors=True)