""" ERGON F8 smoke test — MVP Documentos (Sprint 3 2026-04-20). Valida los 7 criterios de cierre §4.5 del prompt Sesion F: 1. Subir contrato_prueba.pdf con categoria=contrato, tags=fundacion,torre-a → visible en tabla. 2. Descargar desde otra maquina con sesion vitrium: recibe el PDF original. 3. DELETE como cliente → 403. 4. DELETE como admin → 204 + archivo desaparece en disco y DB. 5. Filtro ?tag=fundacion devuelve solo docs con ese tag. 6. Subir el mismo PDF dos veces: segundo upload avisa dedup (misma sha256). 7. Limite 50MB probado con rechazo 413. Uso: python db/smoke_test_f8.py # tests directos DB+filesystem python db/smoke_test_f8.py --with-http # end-to-end con server en :8080 """ from __future__ import annotations import argparse import http.cookiejar import io import json import os import sqlite3 import subprocess import sys import urllib.request from pathlib import Path from urllib.parse import urlencode HERE = Path(__file__).resolve().parent BASE = HERE.parent SERVER_URL = "http://localhost:8080" DOCS_DIR = BASE / "db" / "docs" sys.path.insert(0, str(HERE)) sys.path.insert(0, str(BASE)) import api_obras # type: ignore # noqa: E402 class T: total = 0 passed = 0 failed: list[str] = [] @classmethod def check(cls, name, cond, detail=""): cls.total += 1 if cond: cls.passed += 1 print(f" [OK] {name}" + (f" — {detail}" if detail else "")) return True cls.failed.append(name) print(f" [FAIL] {name}" + (f" — {detail}" if detail else "")) return False @classmethod def summary(cls): print() print(f"Resultados: {cls.passed}/{cls.total} OK") if cls.failed: print("Fallos:") for f in cls.failed: print(f" - {f}") return 1 return 0 def _ensure_vitrium_acceso() -> None: """F13 integracion: asegura que usuario vitrium tiene acceso a VITR1 para tests cliente.""" conn = api_obras.get_conn() try: vit_user = conn.execute( "SELECT id FROM usuarios WHERE email = 'vitrium@demo.local'" ).fetchone() vitr1 = conn.execute("SELECT id FROM obras WHERE codigo = 'VITR1'").fetchone() if vit_user and vitr1: conn.execute( "INSERT OR IGNORE INTO usuarios_obras (usuario_id, obra_id) VALUES (?, ?)", (vit_user["id"], vitr1["id"]), ) conn.commit() finally: conn.close() # --------------------------------------------------------------------------- # DB-level tests (no HTTP) # --------------------------------------------------------------------------- def test_create_list_delete_basico(): """Criterios 1, 5, 6: create + list filtrado + dedup.""" conn = api_obras.get_conn() # Cleanup previo conn.execute( "DELETE FROM documentos WHERE sha256 = ?", ("a" * 64,), ) conn.commit() try: blob_rel = "db/docs/1/sanity-test.pdf" # Asegurar archivo fake existe en disco para el test de delete target = BASE / "db" / "docs" / "1" / "sanity-test.pdf" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"%PDF-1.4\n fake contenido sanity\n%%EOF") res = api_obras.create_documento(conn, "TDG1", { "categoria": "contrato", "titulo": "Contrato sanity test", "tags": "fundacion, torre-a", "sha256": "a" * 64, "mime_type": "application/pdf", "size_bytes": target.stat().st_size, "fecha_doc": "2025-01-15", "subido_por": "smoke@test", }, blob_rel) T.check("1.1 create_documento retorna ok", res.get("ok") is True, f"status={res.get('status')}") doc_id = res.get("documento", {}).get("id") ls = api_obras.list_documentos(conn, "TDG1", categoria="contrato") T.check("1.2 list by categoria contiene el nuevo doc", any(d["id"] == doc_id for d in ls), f"found {len(ls)} docs en contrato") ls_tag = api_obras.list_documentos(conn, "TDG1", tag="fundacion") T.check("5.1 list by tag=fundacion retorna el doc", any(d["id"] == doc_id for d in ls_tag), f"found {len(ls_tag)} con tag fundacion") T.check("5.2 tags normalizados en orden alfabetico", res["documento"]["tags"] == "fundacion,torre-a", f"tags={res['documento']['tags']}") # Dedup: segundo upload mismo sha blob_rel2 = "db/docs/1/sanity-dup.pdf" target2 = BASE / "db" / "docs" / "1" / "sanity-dup.pdf" target2.write_bytes(b"%PDF-1.4 dup %%EOF") res_dup = api_obras.create_documento(conn, "TDG1", { "categoria": "contrato", "titulo": "dup", "sha256": "a" * 64, "mime_type": "application/pdf", "size_bytes": 20, }, blob_rel2) T.check("6. dedup: segundo upload reporta duplicados", isinstance(res_dup.get("duplicados"), list) and len(res_dup["duplicados"]) >= 1, f"duplicados={len(res_dup.get('duplicados') or [])}") # Delete cleanup del_res = api_obras.delete_documento(conn, "TDG1", doc_id) T.check("4.1 delete retorna ok + blob_path", del_res.get("ok") and del_res.get("blob_path") == blob_rel, "") for d in (res_dup.get("documento") and [res_dup["documento"]["id"]] or []): api_obras.delete_documento(conn, "TDG1", d) # Limpiar fakes for p in (target, target2): if p.exists(): p.unlink() finally: conn.close() def test_categoria_invalida(): """Validacion: categoria fuera del enum -> 400.""" conn = api_obras.get_conn() try: res = api_obras.create_documento(conn, "TDG1", { "categoria": "invalida", "titulo": "x", "sha256": "b" * 64, "mime_type": "x", "size_bytes": 1, }, "fake/path.bin") finally: conn.close() T.check("validacion: categoria invalida rechazada con 400", res.get("status") == 400 and not res.get("ok"), f"status={res.get('status')}") def test_titulo_obligatorio(): conn = api_obras.get_conn() try: res = api_obras.create_documento(conn, "TDG1", { "categoria": "contrato", "titulo": "", "sha256": "c" * 64, "mime_type": "x", "size_bytes": 1, }, "fake/path.bin") finally: conn.close() T.check("validacion: titulo vacio rechazado", res.get("status") == 400, f"status={res.get('status')}") def test_sha256_formato(): conn = api_obras.get_conn() try: res = api_obras.create_documento(conn, "TDG1", { "categoria": "contrato", "titulo": "x", "sha256": "abc", "mime_type": "x", "size_bytes": 1, }, "fake/path.bin") finally: conn.close() T.check("validacion: sha256 longitud != 64 rechazado", res.get("status") == 400, f"status={res.get('status')}") # --------------------------------------------------------------------------- # HTTP tests (end-to-end con server live) # --------------------------------------------------------------------------- def _login(opener, email, password): body = json.dumps({"email": email, "password": password}).encode("utf-8") req = urllib.request.Request( SERVER_URL + "/api/auth/login", data=body, headers={"Content-Type": "application/json"}, method="POST", ) with opener.open(req, timeout=10) as r: return r.status def _post_multipart(opener, url, file_bytes, filename, fields): """POST multipart/form-data minimal.""" boundary = "----ergonsmokeboundary" body = bytearray() for k, v in fields.items(): body += f"--{boundary}\r\n".encode() body += f'Content-Disposition: form-data; name="{k}"\r\n\r\n'.encode() body += (v or "").encode("utf-8") + b"\r\n" body += f"--{boundary}\r\n".encode() body += f'Content-Disposition: form-data; name="archivo"; filename="{filename}"\r\n'.encode() body += b"Content-Type: application/pdf\r\n\r\n" body += file_bytes + b"\r\n" body += f"--{boundary}--\r\n".encode() req = urllib.request.Request(url, data=bytes(body), method="POST") req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}") try: with opener.open(req, timeout=60) as r: return r.status, json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: return e.code, json.loads(e.read().decode("utf-8")) def test_http_full_flow(): """Criterios 1, 2, 3, 4, 5, 6, 7 via HTTP end-to-end.""" print() print(" [HTTP] login admin + upload + list + download + delete cliente (403) + delete admin (204)") _ensure_vitrium_acceso() cj_admin = http.cookiejar.CookieJar() op_admin = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj_admin)) try: s = _login(op_admin, "admin@demo.local", "demo1234") T.check("HTTP.0.1 login admin", s == 200, f"status={s}") except Exception as e: T.check("HTTP.0.1 login admin", False, f"error: {e}") return cj_cli = http.cookiejar.CookieJar() op_cli = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj_cli)) try: s = _login(op_cli, "vitrium@demo.local", "vitrium1234") T.check("HTTP.0.2 login cliente", s == 200, f"status={s}") except Exception as e: T.check("HTTP.0.2 login cliente", False, f"error: {e}") return pdf_bytes = b"%PDF-1.4\n%Contrato prueba smoke F8\n%%EOF" url_upload = SERVER_URL + "/api/obras/VITR1/documentos" # 1+5+6: admin sube el doc con tags; list filtra; segundo upload dedup status, data = _post_multipart(op_admin, url_upload, pdf_bytes, "contrato_prueba.pdf", { "categoria": "contrato", "titulo": "Contrato prueba F8", "tags": "fundacion,torre-a", "fecha_doc": "2025-03-15", }) T.check("1. POST upload status 201", status == 201, f"status={status}") doc_id = (data.get("documento") or {}).get("id") T.check("1.2 respuesta tiene documento.id", doc_id is not None, f"doc_id={doc_id}") # 5. Filtro por tag url_filtro = SERVER_URL + "/api/obras/VITR1/documentos?" + urlencode({"tag": "fundacion"}) with op_admin.open(url_filtro, timeout=10) as r: docs_tag = json.loads(r.read().decode("utf-8")) T.check("5. GET ?tag=fundacion contiene el nuevo doc", any(d["id"] == doc_id for d in docs_tag), f"found {len(docs_tag)}") # 6. Dedup: segundo upload mismo contenido -> respuesta incluye dedup.warning status2, data2 = _post_multipart(op_admin, url_upload, pdf_bytes, "contrato_prueba_copia.pdf", { "categoria": "contrato", "titulo": "Copia con otro titulo", }) T.check("6. segundo upload mismo sha -> dedup warning", status2 == 201 and isinstance(data2.get("dedup"), dict) and data2["dedup"].get("warning"), f"status={status2} dedup={data2.get('dedup') is not None}") doc_id_dup = (data2.get("documento") or {}).get("id") # 2. Cliente descarga el blob y recibe el PDF original url_blob = SERVER_URL + f"/api/obras/VITR1/documentos/{doc_id}/blob" try: with op_cli.open(url_blob, timeout=10) as r: blob_data = r.read() cd = r.headers.get("Content-Disposition") or "" mime = r.headers.get("Content-Type") or "" except Exception as e: T.check("2. cliente descarga blob", False, f"error: {e}") return T.check("2.1 cliente recibe bytes identicos al original", blob_data == pdf_bytes, f"len={len(blob_data)}") T.check("2.2 Content-Disposition attachment con filename", "attachment" in cd and ".pdf" in cd, cd) T.check("2.3 Content-Type application/pdf", "pdf" in mime, mime) # 3. DELETE como cliente -> 403 url_del = SERVER_URL + f"/api/obras/VITR1/documentos/{doc_id_dup}" try: req = urllib.request.Request(url_del, method="DELETE") op_cli.open(req, timeout=10) T.check("3. DELETE como cliente -> 403", False, "servidor acepto DELETE de cliente") except urllib.error.HTTPError as e: T.check("3. DELETE como cliente -> 403", e.code == 403, f"status={e.code}") # 4. DELETE como admin -> 204 + archivo borrado de disco try: req = urllib.request.Request(url_del, method="DELETE") with op_admin.open(req, timeout=10) as r: T.check("4.1 DELETE admin -> 204", r.status == 204, f"status={r.status}") except urllib.error.HTTPError as e: T.check("4.1 DELETE admin -> 204", False, f"HTTP {e.code}: {e.read().decode()}") # 4.2 Verificar que el archivo desaparecio de disco (via API: 404 blob) try: with op_admin.open(SERVER_URL + f"/api/obras/VITR1/documentos/{doc_id_dup}", timeout=10) as r: _ = r.read() T.check("4.2 get por id doc borrado -> 404", False, "servidor devolvio 200 para doc borrado") except urllib.error.HTTPError as e: T.check("4.2 get por id doc borrado -> 404", e.code == 404, f"status={e.code}") # 7. Limite 50MB: enviar un archivo 51MB y recibir 413 big_bytes = b"\0" * (51 * 1024 * 1024) status_big, data_big = _post_multipart(op_admin, url_upload, big_bytes, "big.bin", { "categoria": "informe", "titulo": "grande", }) T.check("7. archivo > 50MB rechazado con 413", status_big == 413, f"status={status_big}") # Cleanup: borrar el doc_id original try: req = urllib.request.Request( SERVER_URL + f"/api/obras/VITR1/documentos/{doc_id}", method="DELETE", ) op_admin.open(req, timeout=10) except Exception: pass def main(): parser = argparse.ArgumentParser() parser.add_argument("--with-http", action="store_true") args = parser.parse_args() print(f"ERGON F8 smoke test (base={BASE})") print() test_create_list_delete_basico() test_categoria_invalida() test_titulo_obligatorio() test_sha256_formato() if args.with_http: test_http_full_flow() else: print(" [SKIP] HTTP tests (pasar --with-http con server activo)") sys.exit(T.summary()) if __name__ == "__main__": main()