# -*- coding: utf-8 -*- r""" Prueba de la ventana, con la ventana de verdad. probar.py y prueba_funcional.py ejercitan el programa sin abrirlo; esto abre OrganiLAB sobre una copia temporal (programa y datos aparte, en otro puerto) y comprueba desde otro hilo lo que solo se puede comprobar con la ventana delante: py -3.12 prueba_ventana.py el programa desde el código py -3.12 prueba_ventana.py --paquete dist\OrganizadorLabstream el paquete, arrancado por su .exe … --captura ruta.png deja una imagen de la ventana entera Durante medio minuto se ve una ventana que se abre, se mueve sola y se cierra. No hay que tocar nada. Sale con código 0 si todo pasó. """ import os import sys import json import time import zlib import shutil import socket import struct import ctypes import tempfile import threading import subprocess import urllib.request from ctypes import wintypes as W AQUI = os.path.dirname(os.path.abspath(__file__)) MODULOS = ("organizador.py", "reglas.py", "asistente.py", "actualizador.py", "resolve_bridge.py", "firma.py", "ventana.py", "instalar.py", "WebView2Loader.dll") RES = [] # (nombre, pasó, detalle) U = ctypes.WinDLL("user32") G = ctypes.WinDLL("gdi32") ENUM = ctypes.WINFUNCTYPE(W.BOOL, W.HWND, W.LPARAM) for _n, _a, _r in (("GetWindow", [W.HWND, W.UINT], W.HWND), ("GetClassNameW", [W.HWND, W.LPWSTR, ctypes.c_int], ctypes.c_int), ("GetWindowTextW", [W.HWND, W.LPWSTR, ctypes.c_int], ctypes.c_int), ("IsWindowVisible", [W.HWND], W.BOOL), ("EnumWindows", [ENUM, W.LPARAM], W.BOOL), ("EnumChildWindows", [W.HWND, ENUM, W.LPARAM], W.BOOL), ("SendMessageW", [W.HWND, W.UINT, W.WPARAM, ctypes.c_void_p], ctypes.c_ssize_t), ("PostMessageW", [W.HWND, W.UINT, W.WPARAM, W.LPARAM], W.BOOL), ("GetDlgItem", [W.HWND, ctypes.c_int], W.HWND), ("GetWindowDC", [W.HWND], W.HANDLE), ("ReleaseDC", [W.HWND, W.HANDLE], ctypes.c_int), ("PrintWindow", [W.HWND, W.HANDLE, W.UINT], W.BOOL), ("GetWindowRect", [W.HWND, ctypes.POINTER(W.RECT)], W.BOOL)): getattr(U, _n).argtypes, getattr(U, _n).restype = _a, _r for _n, _a, _r in (("CreateCompatibleDC", [W.HANDLE], W.HANDLE), ("CreateCompatibleBitmap", [W.HANDLE, ctypes.c_int, ctypes.c_int], W.HANDLE), ("SelectObject", [W.HANDLE, W.HANDLE], W.HANDLE), ("DeleteObject", [W.HANDLE], W.BOOL), ("DeleteDC", [W.HANDLE], W.BOOL), ("GetDIBits", [W.HANDLE, W.HANDLE, W.UINT, W.UINT, ctypes.c_void_p, ctypes.c_void_p, W.UINT], ctypes.c_int)): getattr(G, _n).argtypes, getattr(G, _n).restype = _a, _r WM_CLOSE, WM_COMMAND, WM_SETTEXT, WM_GETICON = 0x0010, 0x0111, 0x000C, 0x007F def anota(nombre, ok, detalle=""): RES.append((nombre, bool(ok), str(detalle))) print((" bien " if ok else " MAL ") + nombre + (f" · {detalle}" if detalle else ""), flush=True) def espera(condicion, plazo=15.0, paso=0.1): fin = time.time() + plazo while time.time() < fin: r = condicion() if r: return r time.sleep(paso) return condicion() def puerto_libre(): s = socket.socket() s.bind(("127.0.0.1", 0)) p = s.getsockname()[1] s.close() return p def clase(h): b = ctypes.create_unicode_buffer(200) U.GetClassNameW(h, b, 200) return b.value def texto(h): b = ctypes.create_unicode_buffer(2000) U.GetWindowTextW(h, b, 2000) return b.value def hijos(h): out = [] U.EnumChildWindows(h, ENUM(lambda x, _: out.append(x) or True), 0) return out def dialogo_de(hwnd): """El cuadro de diálogo de Windows abierto sobre esa ventana, si hay alguno.""" out = [] U.EnumWindows(ENUM(lambda x, _: (out.append(x) if clase(x) == "#32770" and U.GetWindow(x, 4) == hwnd and U.IsWindowVisible(x) else None) or True), 0) return out[0] if out else None def marco(hwnd, ruta): """La ventana entera (barra de título incluida), tal como la pinta Windows, a un PNG.""" r = W.RECT() U.GetWindowRect(hwnd, ctypes.byref(r)) ancho, alto = r.right - r.left, r.bottom - r.top dc = U.GetWindowDC(hwnd) mem = G.CreateCompatibleDC(dc) bmp = G.CreateCompatibleBitmap(dc, ancho, alto) G.SelectObject(mem, bmp) U.PrintWindow(hwnd, mem, 2) cab = ctypes.create_string_buffer(struct.pack("I", len(d)) + t + d + struct.pack(">I", zlib.crc32(t + d) & 0xffffffff) with open(ruta, "wb") as fh: fh.write(b"\x89PNG\r\n\x1a\n" + trozo(b"IHDR", struct.pack(">IIBBBBB", ancho, alto, 8, 2, 0, 0, 0)) + trozo(b"IDAT", zlib.compress(bytes(filas), 6)) + trozo(b"IEND", b"")) return ancho, alto def cliente_de_ensayo(tmp): raiz = os.path.join(tmp, "Cliente de ensayo") for rel in ("Rodaje/C0001.MP4", "Entregas/Video final.mp4", "notas.txt"): p = os.path.join(raiz, rel) os.makedirs(os.path.dirname(p), exist_ok=True) with open(p, "wb") as fh: fh.write(os.urandom(2048)) return raiz def marca_de(datos): return (zlib.crc32(os.path.normcase(os.path.abspath(datos)).encode("utf-8")) & 0x7FFFFFFF) or 1 def captura_pedida(): return sys.argv[sys.argv.index("--captura") + 1] if "--captura" in sys.argv else "" # -------------------------------------------------------------------------- # desde el código: el programa corre en este mismo proceso y se le puede preguntar todo # -------------------------------------------------------------------------- def desde_el_codigo(tmp): app = os.path.join(tmp, "app") os.makedirs(app) for f in MODULOS: shutil.copy(os.path.join(AQUI, f), app) shutil.copytree(os.path.join(AQUI, "static"), os.path.join(app, "static")) carpeta = cliente_de_ensayo(tmp) puerto = puerto_libre() os.environ["ORGANIZADOR_PUERTO"] = str(puerto) sys.path.insert(0, app) sys.argv = [os.path.join(app, "organizador.py")] import organizador as O import ventana as V base = f"http://127.0.0.1:{puerto}" def pedir(ruta, datos=None): rq = urllib.request.Request(base + ruta, None if datos is None else json.dumps(datos).encode(), {"Content-Type": "application/json", "Origin": base}) return json.loads(urllib.request.urlopen(rq, timeout=90).read()) def otra_copia(): return subprocess.run([sys.executable, os.path.join(app, "organizador.py")], cwd=app, timeout=40, capture_output=True, text=True, encoding="utf-8", errors="replace") def pruebas(): v = espera(lambda: O.VENTANA if O.VENTANA is not None and O.VENTANA.lista else None, 60) anota("la ventana abre y carga la interfaz", v is not None, "" if v else "no cargó en 60 s") if v is None: os._exit(1) time.sleep(1.5) h = v.hwnd anota("es una ventana de Windows propia, con su título y su icono", clase(h) == V.CLASE and texto(h) == "OrganiLAB" and U.SendMessageW(h, WM_GETICON, 1, None) != 0, f"clase {clase(h)} · «{texto(h)}»") anota("dentro va la interfaz del programa", v.evaluar("ESTADO.version") == O.VERSION, v.evaluar("ESTADO.version")) def identidad(): almacen, out = ctypes.c_void_p(), {} V._dll().s.SHGetPropertyStoreForWindow(h, ctypes.byref(V._guid(V.IID_PROPIEDADES)), ctypes.byref(almacen)) for nombre, pid in (("id", 5), ("orden", 2), ("nombre", 4), ("icono", 3)): clave, pv = V.PROPERTYKEY(V._guid(V.FMT_APPUSERMODEL), pid), V.PROPVARIANT() V._m(almacen.value, 5, ctypes.byref(clave), ctypes.byref(pv), tipos=(ctypes.POINTER(V.PROPERTYKEY), ctypes.POINTER(V.PROPVARIANT))) out[nombre] = ctypes.wstring_at(pv.valor) if pv.vt == 31 and pv.valor else "" V._soltar(almacen.value) return out quien = v.en_ui(identidad) anota("la barra de tareas la conoce como OrganiLAB y sabe cómo volver a abrirla", quien["id"] == V.IDENTIDAD and "organizador.py" in quien["orden"] and quien["nombre"] == "OrganiLAB" and quien["icono"].endswith("organilab.ico"), f"{quien['id']} · {quien['orden']}") def ajustes(): aj = ctypes.c_void_p() V._m(v._wv, 3, ctypes.byref(aj), tipos=(V.PPVOID,)) a3 = V._qi(aj.value, V.IID_AJUSTES3) r = {"barra de estado": V._entero(aj.value, 9), "herramientas de desarrollo": V._entero(aj.value, 11), "zoom": V._entero(aj.value, 17), "página de error": V._entero(aj.value, 19), "teclas de navegador": V._entero(a3, 23)} V._soltar(a3), V._soltar(aj.value) return r aj = v.en_ui(ajustes) anota("lo que es de navegador está apagado", not any(aj.values()), ", ".join(k for k, x in aj.items() if x) or "todo apagado") anota("la interfaz sabe que está en la ventana", v.evaluar("document.documentElement.classList.contains('nativo')") is True) # el avance en la barra de tareas v.progreso(0.5) anota("pone avance en el botón de la barra de tareas", espera(lambda: v.en_ui(lambda: bool(v._barra)), 5)) v.progreso(None) if captura_pedida(): anota("captura de la ventana", True, "%d×%d px en %s" % (*marco(h, captura_pedida()), captura_pedida())) # el selector de carpetas de Windows, pedido como lo pide la interfaz caja = {} threading.Thread(target=lambda: caja.update(r=pedir("/api/elegir-carpeta", {})), daemon=True).start() d = espera(lambda: dialogo_de(h), 15) if d: time.sleep(1.0) campo = U.GetDlgItem(d, 1152) # el campo «Carpeta:» if campo: U.SendMessageW(campo, WM_SETTEXT, 0, ctypes.c_wchar_p(carpeta)) time.sleep(0.3) U.SendMessageW(d, WM_COMMAND, 1, U.GetDlgItem(d, 1)) # con una ruta escrita, el primer «Elegir carpeta» entra en ella y el segundo la elige if not espera(lambda: "r" in caja, 3): U.SendMessageW(d, WM_COMMAND, 1, U.GetDlgItem(d, 1)) if not espera(lambda: "r" in caja, 8): U.PostMessageW(d, WM_CLOSE, 0, 0) espera(lambda: "r" in caja, 8) elegida = (caja.get("r") or {}).get("ruta") anota("el selector de carpetas de Windows se abre sobre la ventana y devuelve la carpeta", bool(d) and elegida and os.path.normcase(elegida) == os.path.normcase(carpeta), elegida or ("se abrió pero no devolvió carpeta" if d else "no se abrió")) # soltar una carpeta: el motor entrega los archivos y la ventana lee su ruta completa def cdp(metodo, parametros): hecho = threading.Event() p, sosten = V._objeto("{5C4889F0-5EF6-4C5A-952C-D8F1B92D0574}", (ctypes.c_long, W.LPCWSTR), lambda hr, r: hecho.set(), fijo=False) v.en_ui(lambda: V._m(v._wv, 36, metodo, json.dumps(parametros), p, tipos=(W.LPCWSTR, W.LPCWSTR, V.PVOID))) hecho.wait(5) return sosten x, y = v.evaluar("[Math.round(innerWidth * 0.6), Math.round(innerHeight * 0.4)]") carga = {"items": [], "files": [carpeta], "dragOperationsMask": 1} for tipo in ("dragEnter", "dragOver", "drop"): cdp("Input.dispatchDragEvent", {"type": tipo, "x": x, "y": y, "data": carga}) llego = espera(lambda: [r for r in O.SOLTADOS["rutas"] if os.path.normcase(r) == os.path.normcase(carpeta)], 8) anota("una carpeta soltada llega con su ruta completa", bool(llego), (llego or ["no llegó"])[0]) propuesta = espera(lambda: v.evaluar("document.querySelector('#modal').classList.contains('on') && !!document.querySelector('#nuevoCliente')"), 8) anota("y la interfaz propone añadirla", propuesta is True, v.evaluar("document.querySelector('#modalTitulo').textContent")) v.evaluar("document.querySelector('#modal').classList.remove('on')") # una sola copia abierta p = otra_copia() anota("abrirlo otra vez no abre una segunda copia", p.returncode == 0 and len(V._ventanas(O._marca())) == 1, f"código {p.returncode} · {len(V._ventanas(O._marca()))} ventana(s)") # cerrar con trabajo en marcha: avisa, se oculta, vuelve si se abre otra vez y se cierra sola al acabar O.APLICANDO.update(total=10, hechas=3, raiz="ensayo") U.PostMessageW(h, WM_CLOSE, 0, 0) d = espera(lambda: dialogo_de(h), 8) aviso = " ".join(texto(x) for x in hijos(d)) if d else "" anota("al cerrar con trabajo en marcha, avisa de qué está haciendo", "moviendo archivos" in aviso, aviso[:90]) if d: U.PostMessageW(d, WM_COMMAND, 1, 0) anota("la ventana se quita de la vista y el programa sigue", espera(lambda: v.oculta and not U.IsWindowVisible(h), 8)) otra_copia() anota("abrirlo otra vez devuelve esa misma ventana", espera(lambda: not v.oculta and U.IsWindowVisible(h), 8)) U.PostMessageW(h, WM_CLOSE, 0, 0) d = espera(lambda: dialogo_de(h), 8) if d: U.PostMessageW(d, WM_COMMAND, 1, 0) espera(lambda: v.oculta, 8) O.APLICANDO.update(hechas=10) # el trabajo termina: el programa tiene que cerrarse solo fin, forzado = threading.Event(), [] def con_red(): try: pruebas() except Exception as ex: # noqa anota("la prueba llegó hasta el final", False, repr(ex)) O.APLICANDO.update(hechas=O.APLICANDO.get("total", 0)) if not fin.wait(10) and O.VENTANA is not None: # no se cerró solo: se cierra a la fuerza para no quedarse aquí forzado.append(1) O.VENTANA.cerrar() threading.Thread(target=con_red, daemon=True).start() t0 = time.time() O.main() fin.set() anota("al terminar el trabajo, el programa se cierra solo", not forzado, f"{time.time() - t0:.0f} s en total") datos = os.path.join(app, "datos") try: sitio = json.load(open(os.path.join(datos, "ventana.json"), encoding="utf-8")) except Exception: # noqa sitio = {} anota("recuerda dónde y de qué tamaño quedó", sitio.get("ancho", 0) > 200, json.dumps(sitio)) anota("no deja rastro de la sesión", not os.path.exists(os.path.join(datos, "sesion.json"))) # -------------------------------------------------------------------------- # el paquete: se arranca por su .exe, como lo arrancaría una persona, y se mira desde fuera # -------------------------------------------------------------------------- def el_paquete(tmp, origen): sys.path.insert(0, AQUI) import ventana as V app = os.path.join(tmp, "OrganizadorLabstream") shutil.copytree(origen, app, ignore=shutil.ignore_patterns("ffmpeg", "datos", "actualizacion", "__pycache__")) exe = os.path.join(app, "OrganizadorLabstream.exe") puerto = puerto_libre() entorno = dict(os.environ, ORGANIZADOR_PUERTO=str(puerto)) base = f"http://127.0.0.1:{puerto}" p = subprocess.run([exe, "-B", "-c", "import sys; print(sys.version_info[:2])"], capture_output=True, text=True, timeout=60, stdin=subprocess.DEVNULL) anota("el .exe del paquete arranca en este Windows", p.returncode == 0, (p.stdout or p.stderr).strip()[:80]) proc = subprocess.Popen([exe], cwd=tmp, env=entorno) def estado(): try: return json.loads(urllib.request.urlopen(base + "/api/estado", timeout=2).read()) except Exception: # noqa return None e = espera(estado, 40, 0.3) anota("doble clic en el .exe abre el programa", bool(e), f"versión {e['version']}" if e else "no respondió en 40 s") if not e: proc.kill() return anota("corre con el motor de su propia carpeta", e.get("congelado") is True) marca = marca_de(os.path.join(app, "datos")) hs = espera(lambda: V._ventanas(marca), 30) h = hs[0] if hs else None anota("con su ventana de Windows", bool(h) and texto(h) == "OrganiLAB" and bool(U.IsWindowVisible(h)), f"clase {clase(h)}" if h else "sin ventana") if h: time.sleep(4) if captura_pedida(): anota("captura de la ventana", True, "%d×%d px en %s" % (*marco(h, captura_pedida()), captura_pedida())) subprocess.run([exe], cwd=tmp, env=entorno, timeout=40) anota("abrirlo otra vez no abre una segunda copia", len(V._ventanas(marca)) == 1, f"{len(V._ventanas(marca))} ventana(s)") U.PostMessageW(h, WM_CLOSE, 0, 0) try: proc.wait(25) except subprocess.TimeoutExpired: proc.kill() anota("al cerrar la ventana se cierra el programa", proc.returncode == 0, f"código {proc.returncode}") anota("y deja de atender", estado() is None) p = subprocess.run([exe, "ayuda"], capture_output=True, timeout=60, stdin=subprocess.DEVNULL, env=entorno) anota("por consola responde a sus comandos", p.returncode == 0 and b"asistente" in p.stdout) def main(): guardia = threading.Timer(300, lambda: os._exit(3)) # pase lo que pase, no se queda colgada guardia.daemon = True guardia.start() tmp = tempfile.mkdtemp(prefix="organilab-ventana-") try: if "--paquete" in sys.argv: el_paquete(tmp, os.path.abspath(sys.argv[sys.argv.index("--paquete") + 1])) else: desde_el_codigo(tmp) finally: shutil.rmtree(tmp, ignore_errors=True) mal = [n for n, ok, _ in RES if not ok] print(f"\n{len(RES) - len(mal)} de {len(RES)} comprobaciones pasan" + (": fallan " + "; ".join(mal) if mal else "")) return 1 if mal else 0 if __name__ == "__main__": sys.exit(main())