labstream-organizador/prueba_funcional.py

# -*- 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
    return 'reconoce, aprende y olvida un nombre de carpeta'


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)