1.6.2: manifiesto firmado, tiempo estimado en tareas, pruebas automáticas
- firma.py: RSA-2048 PKCS#1 v1.5 SHA-256 sin dependencias; empaquetar.py firma el manifiesto y actualizador.py exige la firma (CLAVE_PUBLICA incrustada). - Barra de tareas: tiempo restante estimado a partir del avance. - Corregido: APLICANDO nunca llegaba al total y «Moviendo» quedaba fijo, bloqueando desconectar la carpeta. - probar.py: reglas, firma y recorrido completo sobre una copia temporal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Este commit está contenido en:
Labstream
216
probar.py
Archivo normal
216
probar.py
Archivo normal
@@ -0,0 +1,216 @@
|
||||
# -*- 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_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())
|
||||
# 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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Referencia en una nueva incidencia
Block a user