ad6efef792
Las tres propuestas que faltaban del lienzo: - La cola respeta un horario (espera entre clip y clip) y espera si Resolve, Premiere o After Effects están usando la tarjeta (nvidia-smi pmon). La pasada automática puede meter en cola lo de más de N meses hasta un tope de GB. Al acabar, un aviso de Windows con la identidad del programa. - Cada tanda deja un resumen en la bitácora (op tanda): clips, pesos, tiempo, nota, fallos, por carpeta. Sale en Convertir con «Exportar CSV» y «Abrir la carpeta», y en Historial cuenta clips. - Barra de marcar con criterio (tamaño, cámara, mes, antigüedad) y buscador. Pruebas: probar 19, prueba_orden 33, prueba_funcional 32 (nueva: resumen de la tanda), prueba_ventana 18. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
596 líneas
31 KiB
Python
596 líneas
31 KiB
Python
# -*- 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', 'braw.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)
|
|
|
|
# dos videos de verdad para el convertidor (6 s, 1280x720, H.264): uno con tanto grano y tanta tasa que sí
|
|
# adelgaza, y otro ya comprimido como sale de una cámara, que el convertidor no debe ofrecer
|
|
FF = shutil.which('ffmpeg')
|
|
CLIP = os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'C0020.MP4')
|
|
CLIP2 = os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'C0021.MP4')
|
|
if FF:
|
|
for destino, imagen, crf in ((CLIP, 'testsrc=duration=6:size=1280x720:rate=25,noise=alls=40:allf=t', '8'),
|
|
(CLIP2, 'testsrc=duration=6:size=1280x720:rate=25', '30')):
|
|
subprocess.run([FF, '-hide_banner', '-loglevel', 'error', '-y', '-f', 'lavfi', '-i', imagen,
|
|
'-f', 'lavfi', '-i', 'sine=frequency=440:duration=6', '-c:v', 'libx264', '-preset', 'ultrafast', '-crf', crf,
|
|
'-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', destino], check=True)
|
|
t = time.mktime(time.struct_time((2025, 6, 10, 12, 0, 0, 0, 0, -1)))
|
|
os.utime(destino, (t, t))
|
|
# y, si se da uno, un BRAW de verdad (variable ORGANILAB_BRAW_PRUEBA): los de arriba son bytes al azar
|
|
BRAW = os.environ.get('ORGANILAB_BRAW_PRUEBA') or ''
|
|
if BRAW:
|
|
shutil.copy(BRAW, os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'A001_03051200_C003.braw'))
|
|
t = time.mktime(time.struct_time((2026, 3, 5, 12, 0, 0, 0, 0, -1)))
|
|
os.utime(os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'A001_03051200_C003.braw'), (t, t))
|
|
|
|
# ---------------------------------------------------------------- el servidor
|
|
with socket.socket() as s_:
|
|
s_.bind(('127.0.0.1', 0))
|
|
PUERTO = s_.getsockname()[1]
|
|
ENSAYO = os.path.join(TMP, 'equipo')
|
|
os.makedirs(ENSAYO)
|
|
env = dict(os.environ, ORGANIZADOR_PUERTO=str(PUERTO), PYTHONIOENCODING='utf-8', ORGANILAB_ENSAYO=ENSAYO)
|
|
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']
|
|
av = get('/api/aplicando')
|
|
assert av['lote'] == r['lote'] and av['fin'] and av['unidades'] == av['unidades_total'] > 0 and av['movidos'] == r['movimientos'], av
|
|
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»; el avance acabó en %d de %d' % (r['movimientos'], av['unidades'], av['unidades_total'])
|
|
|
|
|
|
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
|
|
assert 'ok' in (m.get('braw') or {}), m.get('braw')
|
|
assert get('/api/estado')['herramientas']['braw']['ok'] in (True, False)
|
|
return 'formatos disponibles: ' + ', '.join(disp) + ' · GPU: ' + (m['gpu'].get('nombre') or 'no') + \
|
|
' · lector de BRAW: ' + ('sí' if m['braw']['ok'] else 'no (' + (m['braw'].get('motivo') or '')[:40] + ')')
|
|
|
|
|
|
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']
|
|
comprimido = [x for x in cand if x['ruta'].lower().endswith('c0021.mp4')]
|
|
assert comprimido and comprimido[0]['excluido'].startswith('ya viene comprimido (H.264'), (comprimido or cand)[:1]
|
|
return 'leyó %d clip(s) · C0020.MP4 %s %s a %d kb/s: se ofrece · C0021.MP4 a %d kb/s: no' % (
|
|
len(cand), clip[0]['codec'], clip[0]['res'], clip[0]['kbps'], comprimido[0]['kbps'])
|
|
|
|
|
|
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_comparar():
|
|
# el original ya se reemplazó (está en _Para borrar/originales): el cuadro tiene que salir de ahí
|
|
for cual in ('original', 'convertido'):
|
|
r = urllib.request.urlopen(BASE + '/api/optimizar/cuadro' + q(id=IDS['trabajo'], cual=cual, t=1.5), timeout=120)
|
|
assert r.headers.get('Content-Type') == 'image/jpeg' and len(r.read()) > 2000, 'no salió el cuadro %s' % cual
|
|
c, r = post('/api/optimizar/calidad', {'id': IDS['trabajo']})
|
|
assert c == 200 and 30 < r['vmaf'] <= 100 and r['cuadros'] > 10, r
|
|
t = next(x for x in get('/api/optimizar/cola')['trabajos'] if x['id'] == IDS['trabajo'])
|
|
assert t['nota'] == r['vmaf'] and t.get('dur'), t
|
|
return 'dos cuadros JPEG y nota VMAF %s sobre %d cuadros (%.0f s)' % (r['vmaf'], r['cuadros'], r['seg'])
|
|
|
|
|
|
def p_tanda():
|
|
# al acabar la cola queda el resumen de la tanda: clips, pesos, tiempo; y se exporta en CSV
|
|
u = get('/api/optimizar/cola')['cola']['ultima']
|
|
assert u and u['clips'] >= 1 and u['bytes_in'] > u['bytes_out'] > 0 and u['lote'].startswith('tanda-'), u
|
|
assert IDS['trabajo'] in u['ids'] and u['raiz'].lower() == RAIZ.lower(), u
|
|
r = urllib.request.urlopen(BASE + '/api/optimizar/tanda.csv' + q(lote=u['lote']), timeout=60)
|
|
csv_ = r.read().decode('utf-8-sig')
|
|
assert r.headers.get('Content-Type', '').startswith('text/csv') and csv_.startswith('clip;carpeta;estado') and 'reemplazado' in csv_, csv_[:200]
|
|
lotes = get('/api/bitacora')['lotes']
|
|
l = next((x for x in lotes if x['lote'] == u['lote']), None)
|
|
assert l and l['tanda']['clips'] == u['clips'], l
|
|
# el horario: fuera de él la cola espera (se comprueba la regla, no esperando de verdad)
|
|
c, r2 = post('/api/config', {'optimizar': {'horario': {'activo': True, 'desde': '20:00', 'hasta': '07:00'}, 'pausar_si_edito': False, 'avisar': False}})
|
|
assert c == 200 and r2['config']['optimizar']['horario']['desde'] == '20:00'
|
|
c, r3 = post('/api/config', {'optimizar': {'horario': {'activo': False, 'desde': '20:00', 'hasta': '07:00'}}})
|
|
assert c == 200 and not r3['config']['optimizar']['horario']['activo']
|
|
return 'resumen de %d clip(s), %s, CSV con %d líneas, y en Historial' % (u['clips'], u['lote'], csv_.count('\n'))
|
|
|
|
|
|
def p_braw():
|
|
m = get('/api/optimizar/motores')['braw']
|
|
if not BRAW:
|
|
return 'sin clip de prueba (ORGANILAB_BRAW_PRUEBA): solo se comprueba que el lector responde (%s)' % ('sí' if m['ok'] else 'no')
|
|
assert m['ok'], m
|
|
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('a001_03051200_c003.braw')]
|
|
assert clip and clip[0]['motor'] == 'braw' and not clip[0]['excluido'], (clip or cand)[:1]
|
|
c, r = post('/api/optimizar/encolar', {'raiz': RAIZ, 'rutas': [clip[0]['ruta']], 'perfil': 'equilibrado', 'formato': 'h265_10bit', 'chroma': '420', 'color': 'rec709', 'iniciar': True})
|
|
assert c == 200 and r['ids'], r
|
|
assert esperar(lambda: not get('/api/estado')['cola']['activa'], 240), 'la cola no terminó'
|
|
t = next(x for x in get('/api/optimizar/cola' + q(raiz=RAIZ))['trabajos'] if x['id'] == r['ids'][0])
|
|
assert t['estado'] == 'verificado' and t['motor'] == 'braw', (t['estado'], t['motor'], (t.get('log') or '')[-300:])
|
|
assert os.path.isfile(t['salida']) and os.path.isfile(clip[0]['ruta']) # el original sigue en su sitio
|
|
c, rr = post('/api/optimizar/reemplazar', {'ids': r['ids']})
|
|
assert rr['resultados'][0]['ok'], rr
|
|
assert os.path.isfile(t['salida']) and not os.path.isfile(clip[0]['ruta'])
|
|
return 'BRAW %s leído sin Resolve, convertido (%s), verificado y reemplazado' % (clip[0]['res'], (t.get('log') or '').split(' · ')[3][:40])
|
|
|
|
|
|
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_pieza():
|
|
hoy = time.localtime()
|
|
v = get('/api/pieza/previa' + q(raiz=RAIZ, titulo='Reel de prueba', anio=hoy.tm_year, mes=hoy.tm_mon))
|
|
assert v['rel'].endswith('Reel de prueba') and not v['existe'], v
|
|
assert not os.path.exists(v['ruta']), 'la vista previa creó la carpeta'
|
|
c, r = post('/api/pieza', {'raiz': RAIZ, 'titulo': 'Reel de prueba', 'anio': hoy.tm_year, 'mes': hoy.tm_mon})
|
|
assert c == 200 and r.get('ok'), r
|
|
assert sorted(os.listdir(r['ruta'])) == sorted(r['subcarpetas']) and len(r['subcarpetas']) >= 5, os.listdir(r['ruta'])
|
|
c2, r2 = post('/api/pieza', {'raiz': RAIZ, 'titulo': 'reel de prueba', 'anio': hoy.tm_year, 'mes': hoy.tm_mon})
|
|
assert c2 == 400 and 'ya existe' in r2.get('error', ''), (c2, r2)
|
|
vac = [x['rel'] for x in get('/api/vacias' + q(raiz=RAIZ))['carpetas']]
|
|
assert not any('Reel de prueba' in x for x in vac), 'ofrece borrar una pieza recién creada: %s' % vac
|
|
c3, r3 = post('/api/deshacer', {'lote': r['lote']})
|
|
assert c3 == 200 and not r3['errores'] and not os.path.exists(r['ruta']), r3
|
|
return 'crea «%s» con %d subcarpetas, no repite, y se deshace' % (r['nombre'], len(r['subcarpetas']))
|
|
|
|
|
|
def p_entrada():
|
|
tarjeta = os.path.join(TMP, 'tarjeta')
|
|
clip = os.path.join(tarjeta, 'DCIM', '100MEDIA', 'DJI_0007.MP4')
|
|
os.makedirs(os.path.dirname(clip))
|
|
with open(clip, 'wb') as fh:
|
|
fh.write(os.urandom(700_000))
|
|
t = time.mktime(time.struct_time((2026, 8, 10, 12, 0, 0, 0, 0, -1)))
|
|
os.utime(clip, (t, t))
|
|
c, p = post('/api/entrada/plan', {'raiz': RAIZ, 'origen': tarjeta})
|
|
assert c == 200 and p['n'] == 1 and p['bytes'] == 700_000, p
|
|
c, mal = post('/api/entrada/plan', {'raiz': RAIZ, 'origen': os.path.join(RAIZ, '2026')})
|
|
assert c == 400 and mal.get('error'), mal
|
|
c, r = post('/api/entrada/traer', {'raiz': RAIZ, 'origen': tarjeta})
|
|
assert c == 200 and r.get('lanzado'), r
|
|
assert esperar(lambda: get('/api/estado')['traer'].get('estado') not in ('copiando', 'ordenando'), 120)
|
|
tr = get('/api/estado')['traer']
|
|
assert tr['estado'] == 'listo' and tr['copiados'] == 1 and not tr['errores_n'], tr
|
|
final = os.path.join(RAIZ, '2026', '08- Agosto', '00_Grabacion', 'DJI', 'DJI_0007.MP4')
|
|
assert os.path.isfile(final), 'no llegó a su sitio: %s' % [x for x in arbol() if 'dji' in x]
|
|
assert os.path.getsize(final) == 700_000 and abs(os.path.getmtime(final) - t) < 2, 'la copia no conserva peso o fecha'
|
|
assert os.path.isfile(clip), 'tocó la tarjeta'
|
|
e = get('/api/entrada' + q(raiz=RAIZ))
|
|
assert e['n'] == 0 and e['existe'], e
|
|
# y lo que se deja a mano en la bandeja sale en la lista y se reparte con su botón
|
|
suelto = os.path.join(RAIZ, '_Entrada', 'C0099.MP4')
|
|
with open(suelto, 'wb') as fh:
|
|
fh.write(os.urandom(4096))
|
|
os.utime(suelto, (t, t))
|
|
a = get('/api/carpeta/acciones' + q(raiz=RAIZ))
|
|
assert any(x['clave'] == 'entrada' and x['estado'] == 'toca' for x in a['acciones']), [x['clave'] for x in a['acciones']]
|
|
c, o = post('/api/entrada/ordenar', {'raiz': RAIZ})
|
|
assert c == 200 and o['movimientos'] >= 1 and not o['errores'] and o['quedan'] == 0, o
|
|
assert os.path.isfile(os.path.join(RAIZ, '2026', '08- Agosto', '00_Grabacion', 'SONY', 'C0099.MP4')), [x for x in arbol() if 'c0099' in x]
|
|
return 'trae de una tarjeta con copia comprobada y la reparte; lo dejado a mano en la bandeja también'
|
|
|
|
|
|
def p_automatico():
|
|
c, r = post('/api/automatico/config', {'activo': True, 'hora': '03:15'})
|
|
assert c == 200 and r['activo'] and r['hora'] == '03:15', r
|
|
tarea = json.load(open(os.path.join(ENSAYO, 'tarea.json'), encoding='utf-8'))
|
|
assert tarea['activo'] and tarea['orden'].endswith(' automatico'), tarea
|
|
c, mal = post('/api/automatico/config', {'activo': True, 'hora': 'a las tres'})
|
|
assert c == 400, mal
|
|
c, r = post('/api/automatico/ahora', {})
|
|
assert c == 200 and r.get('lanzado'), r
|
|
assert esperar(lambda: not get('/api/estado')['automatico'].get('activo'), 180)
|
|
u = get('/api/automatico')['ultimo']
|
|
assert u and len(u['carpetas']) == 1 and u['carpetas'][0]['estado'] in ('al dia', 'con dudas'), u
|
|
assert not u['carpetas'][0]['errores'], u['carpetas'][0]
|
|
# lo que lanza la tarea de Windows cuando el programa ya está abierto: se lo pide a él y se va, para
|
|
# que no haya dos procesos moviendo lo mismo
|
|
r = subprocess.run([sys.executable, 'organizador.py', 'automatico'], cwd=APP, env=env, timeout=120)
|
|
assert r.returncode == 0, 'la orden «automatico» salió con %s' % r.returncode
|
|
assert esperar(lambda: not get('/api/estado')['automatico'].get('activo'), 180)
|
|
u2 = get('/api/automatico')['ultimo']
|
|
assert u2['origen'] == 'programado' and u2['ts'] > u['ts'], u2
|
|
c, r = post('/api/automatico/config', {'activo': False, 'hora': '03:15'})
|
|
assert c == 200 and not r['activo'], r
|
|
return 'se enciende (tarea de Windows en ensayo), hace una pasada y deja el resumen (%s); con el programa abierto, la tarea se lo pide a él' % u['carpetas'][0]['estado']
|
|
|
|
|
|
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), ('Comparar original y convertido', p_comparar), ('Resumen de la tanda', p_tanda), ('Convertir un BRAW', p_braw), ('Control de la cola', p_cola_control),
|
|
('Respaldar', p_respaldo), ('Asistente / hacer de una vez', p_asistente),
|
|
('Pieza nueva', p_pieza), ('Bandeja de entrada y tarjeta', p_entrada), ('Orden automático', p_automatico),
|
|
('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)
|