dataforge/fn_registry
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat: cierra issues 0050 y 0052 + commands automáticos

Browse Source 
- 0050: jupyter_exec reescrito sin Y.js (REST + KernelClient). Bug raíz adicional: HEAD /api/contents da 405 → cambiado a GET. 9 tests (5 unit + 4 e2e).
- 0052: footprint_aurgi cerrado. Bug fix en setup_geo_stack_docker_pipeline (verify aborta si compose up falla; nombre de contenedor incorrecto).
- Nueva primitiva docker_container_running_py_infra (7 tests).
- /full-git-push y /full-git-pull pasan a modo automático: auto-commit + push sin preguntar, aborta solo si detecta secrets.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Egutierrez
2026-05-05 23:34:03 +02:00
parent 1e8ade0ed4
commit 5194de3c04
14 changed files with 684 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/functions/notebook/jupyter_exec.md
+32 -30
View File
@@ -3,17 +3,17 @@ name: jupyter_exec
kind: function
lang: py
domain: notebook
version: "1.0.0"
version: "2.0.0"
purity: impure
signature: "jupyter_append_execute(notebook_path: str, code: str, server_url: str, token: str) -> dict"
description: "Ejecuta codigo en kernels de Jupyter via WebSocket. Tres modos: append (añade celda al notebook y la ejecuta), cell (ejecuta celda existente por indice), kernel (ejecuta en el kernel sin tocar ningun notebook)."
description: "Ejecuta codigo en kernels de Jupyter via REST + WebSocket clasico al kernel. Tres modos: append (añade celda y ejecuta), cell (ejecuta celda existente), kernel (ejecuta sin tocar notebook). NO usa el canal colaborativo Y.js."
tags: [jupyter, notebook, kernel, websocket, execution, cells]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [jupyter_kernel_client, jupyter_nbmodel_client]
imports: [jupyter_kernel_client, urllib, json, uuid]
params:
- name: notebook_path
desc: "Ruta relativa al notebook"
@@ -24,9 +24,18 @@ params:
- name: token
desc: "Token de autenticación (default vacío)"
output: "Dict con cell_index y outputs del código ejecutado, o resultados del kernel"
tested: false
tests: []
test_file_path: ""
tested: true
tests:
- "test_notebook_exists_uses_get_not_head"
- "test_notebook_exists_returns_false_on_404"
- "test_create_notebook_skips_when_exists"
- "test_new_code_cell_has_required_fields"
- "test_extract_outputs_handles_streams_and_results"
- "e2e: test_e2e_append_executes_and_persists"
- "e2e: test_e2e_append_twice_increments_index"
- "e2e: test_e2e_cell_executes_existing"
- "e2e: test_e2e_kernel_mode"
test_file_path: "python/functions/notebook/tests/test_jupyter_exec.py"
file_path: "python/functions/notebook/jupyter_exec.py"
---
@@ -34,9 +43,9 @@ file_path: "python/functions/notebook/jupyter_exec.py"
### `jupyter_append_execute(notebook_path, code, server_url, token)`
Añade una celda de codigo al final del notebook y la ejecuta. Usa el protocolo
colaborativo de Jupyter, por lo que tanto el agente como el usuario ven la celda
y su output en tiempo real en JupyterLab.
Añade una celda de codigo al final del notebook, la ejecuta en el kernel y persiste
celda + outputs a disco via REST `/api/contents`. Jupyter Lab detecta el cambio y lo
refleja en el browser.
```python
from notebook.jupyter_exec import jupyter_append_execute
@@ -52,23 +61,18 @@ result = jupyter_append_execute(
### `jupyter_execute_cell(notebook_path, cell_index, server_url, token)`
Ejecuta una celda existente del notebook por su indice (0-based).
Ejecuta una celda existente por indice (0-based) y persiste sus outputs.
```python
from notebook.jupyter_exec import jupyter_execute_cell
result = jupyter_execute_cell("notebooks/analisis.ipynb", 3)
# {"cell_index": 3, "outputs": ["42"]}
```
### `jupyter_kernel_execute(code, server_url, token)`
Ejecuta codigo directamente en el kernel sin modificar ningun notebook. Util para
consultas rapidas, inspeccion de variables o verificacion de estado del kernel.
Ejecuta codigo directo en el kernel sin tocar ningun notebook.
```python
from notebook.jupyter_exec import jupyter_kernel_execute
result = jupyter_kernel_execute("len(df)")
# {"outputs": ["1500"], "status": "ok"}
```
@@ -76,13 +80,8 @@ result = jupyter_kernel_execute("len(df)")
## CLI
```bash
# Añadir celda y ejecutar
python -m notebook.jupyter_exec append notebooks/mi.ipynb "print('hola')" --server http://localhost:8888 --token mytoken
# Ejecutar celda existente
python -m notebook.jupyter_exec cell notebooks/mi.ipynb 2 --server http://localhost:8888
# Ejecutar en kernel directamente
python -m notebook.jupyter_exec append notebooks/mi.ipynb "print('hola')"
python -m notebook.jupyter_exec cell notebooks/mi.ipynb 2
python -m notebook.jupyter_exec kernel "x = 42; print(x)"
```
@@ -96,12 +95,15 @@ Output siempre JSON. En error retorna `{"error": "..."}` por stderr con exit cod
| display_data / execute_result | `data.text/plain` |
| error | `traceback` (joined con `\n`) |
## Notas
## Notas (v2.0.0 — fix Issue 0050)
- Las funciones `append` y `cell` son async internamente; las publicas usan `asyncio.run()`.
- `jupyter_kernel_execute` es sincrona directamente porque `KernelClient.execute` es bloqueante.
- **Bypassa el canal colaborativo Y.js**. Usa REST `/api/contents` para leer/escribir
celdas y `KernelClient` (websocket clasico al kernel) para ejecutar. Robusto frente
a versiones nuevas de `jupyter-collaboration` que rompian `NbModelClient`.
- **Trade-off**: las celdas/outputs se persisten a disco, no se sincronizan en
tiempo real via Y.js. Jupyter Lab detecta el cambio en el filesystem y lo refleja
(puede pedir 'Revert to disk' segun version).
- `_notebook_exists` usa `GET /api/contents?content=0` (HEAD devuelve 405 en Jupyter Server).
- **Auto-init**: `jupyter_append_execute` crea el notebook si no existe y arranca una
sesion con kernel si no hay ninguna activa para ese notebook.
- El token puede ser cadena vacia si el servidor tiene autenticacion deshabilitada.
- `NbModelClient` requiere que el servidor tenga habilitado el endpoint colaborativo (`/api/collaboration/`), disponible en JupyterLab >= 4 con `jupyter-collaboration` instalado.
- **Auto-init**: `jupyter_append_execute` crea el notebook automaticamente si no existe (via REST PUT /api/contents) y arranca una sesion con kernel si no hay ninguna activa para ese notebook (via POST /api/sessions). No es necesario abrir el notebook manualmente en el navegador.
- **Auto-session**: `jupyter_execute_cell` tambien garantiza que exista una sesion con kernel antes de ejecutar.
- **Fix Issue 006**: `jupyter_execute_cell` normaliza la celda antes de ejecutar. Las celdas creadas manualmente (no via la UI de Jupyter) pueden carecer de `outputs` o `execution_count` en el modelo CRDT, lo que causaba `KeyError: 'outputs'` dentro de `execute_cell` al hacer `del ycell["outputs"][:]`. El fix lee la celda con `nb[cell_index]`, detecta los campos faltantes, y reemplaza la celda via `nb[cell_index] = _normalize_code_cell(cell)` — que usa `set_cell` internamente para re-crear el mapa CRDT completo preservando el source original.
python/functions/notebook/jupyter_exec.py
+119 -198
View File
@@ -1,35 +1,48 @@
"""Ejecuta codigo en kernels de Jupyter via WebSocket.
"""Ejecuta codigo en kernels de Jupyter.
Tres modos de ejecucion:
Tres modos:
- append: añade una celda al final del notebook y la ejecuta
- cell: ejecuta una celda existente por indice
- kernel: ejecuta codigo directamente en el kernel sin modificar ningun notebook
- cell: ejecuta una celda existente por indice
- kernel: ejecuta codigo directamente en el kernel sin tocar notebook
Implementacion basada en REST `/api/contents` + `KernelClient` (websocket clasico
al kernel). NO usa `jupyter_nbmodel_client` ni el canal colaborativo Y.js, por lo
que es robusto frente a versiones nuevas de `jupyter-collaboration` (ver issue
0050). Trade-off: los cambios al notebook se persisten a disco; Jupyter Lab los
detecta via file watch (puede pedir 'Revert to disk' o 'Overwrite' segun version).
"""
import asyncio
import json
from functools import partial
import uuid
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from jupyter_kernel_client import KernelClient
from jupyter_nbmodel_client import NbModelClient, get_jupyter_notebook_websocket_url
from nbformat import NotebookNode
# ---------------------------------------------------------------------------
# Helpers internos
# Helpers REST
# ---------------------------------------------------------------------------
def _auth_headers(token: str, content_type: bool = False) -> dict[str, str]:
headers = {"Accept": "application/json"}
if content_type:
headers["Content-Type"] = "application/json"
if token:
headers["Authorization"] = f"token {token}"
return headers
def _notebook_exists(notebook_path: str, server_url: str, token: str) -> bool:
"""Comprueba si un notebook existe en el servidor Jupyter via HEAD /api/contents."""
headers = {"Accept": "application/json"}
if token:
headers["Authorization"] = f"token {token}"
check_url = f"{server_url}/api/contents/{notebook_path}"
req = Request(check_url, headers=headers, method="HEAD")
"""Comprueba si un notebook existe via GET /api/contents (con `content=0`).
Nota: Jupyter Server no soporta HEAD en /api/contents (responde 405). Usamos
GET con content=0 para evitar transferir el cuerpo completo.
"""
check_url = f"{server_url}/api/contents/{notebook_path}?content=0"
req = Request(check_url, headers=_auth_headers(token), method="GET")
try:
with urlopen(req, timeout=5):
return True
@@ -43,12 +56,6 @@ def _create_notebook(notebook_path: str, server_url: str, token: str, kernel_nam
"""Crea un notebook vacio via PUT /api/contents si no existe."""
if _notebook_exists(notebook_path, server_url, token):
return
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
if token:
headers["Authorization"] = f"token {token}"
kernel_display = {"python3": "Python 3 (ipykernel)", "python": "Python 3"}.get(kernel_name, kernel_name)
notebook_content = {
"nbformat": 4,
@@ -61,49 +68,53 @@ def _create_notebook(notebook_path: str, server_url: str, token: str, kernel_nam
}
body = json.dumps({"type": "notebook", "content": notebook_content}).encode("utf-8")
url = f"{server_url}/api/contents/{notebook_path}"
req = Request(url, data=body, headers=headers, method="PUT")
req = Request(url, data=body, headers=_auth_headers(token, content_type=True), method="PUT")
with urlopen(req, timeout=10) as resp:
resp.read()
def _get_notebook_content(notebook_path: str, server_url: str, token: str) -> dict:
"""Lee el notebook completo via GET /api/contents (con `content`)."""
url = f"{server_url}/api/contents/{notebook_path}?content=1&type=notebook"
req = Request(url, headers=_auth_headers(token), method="GET")
with urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
def _put_notebook_content(notebook_path: str, server_url: str, token: str, content: dict) -> None:
"""Sobrescribe el notebook via PUT /api/contents."""
body = json.dumps({"type": "notebook", "format": "json", "content": content}).encode("utf-8")
url = f"{server_url}/api/contents/{notebook_path}"
req = Request(url, data=body, headers=_auth_headers(token, content_type=True), method="PUT")
with urlopen(req, timeout=10) as resp:
resp.read()
def _ensure_session(server_url: str, token: str, notebook_path: str, kernel_name: str = "python3") -> str:
"""Garantiza que exista una sesion para el notebook. Retorna el kernel_id.
"""Garantiza una sesion para el notebook. Retorna kernel_id.
Si ya hay una sesion activa, retorna su kernel_id. Si no, crea una nueva
via POST /api/sessions (lo cual tambien arranca un kernel).
Si existe una sesion vinculada al notebook, reusa su kernel. Si no, crea
sesion+kernel via POST /api/sessions.
"""
kernel_id = _resolve_kernel_id(server_url, token, notebook_path)
if kernel_id:
return kernel_id
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
}
if token:
headers["Authorization"] = f"token {token}"
body = json.dumps({
"path": notebook_path,
"type": "notebook",
"kernel": {"name": kernel_name},
}).encode("utf-8")
url = f"{server_url}/api/sessions"
req = Request(url, data=body, headers=headers, method="POST")
req = Request(url, data=body, headers=_auth_headers(token, content_type=True), method="POST")
with urlopen(req, timeout=10) as resp:
session = json.loads(resp.read())
return session.get("kernel", {}).get("id", "")
def _api_get(url: str, token: str = "") -> dict | list | None:
"""GET a Jupyter REST API endpoint."""
headers = {"Accept": "application/json"}
if token:
headers["Authorization"] = f"token {token}"
try:
req = Request(url, headers=headers)
req = Request(url, headers=_auth_headers(token))
with urlopen(req, timeout=5) as resp:
return json.loads(resp.read())
except (URLError, OSError, json.JSONDecodeError):
@@ -111,7 +122,7 @@ def _api_get(url: str, token: str = "") -> dict | list | None:
def _resolve_kernel_id(server_url: str, token: str, notebook_path: str) -> str | None:
"""Find the kernel_id associated with a notebook via the sessions API."""
"""Busca el kernel_id de la sesion del notebook via /api/sessions."""
sessions = _api_get(f"{server_url}/api/sessions", token) or []
for session in sessions:
nb = session.get("notebook", session.get("path", {}))
@@ -122,34 +133,20 @@ def _resolve_kernel_id(server_url: str, token: str, notebook_path: str) -> str |
return None
def _resolve_collab_username(server_url: str, token: str) -> str:
"""Resolve the display name of the active user in Jupyter collaboration.
Queries /api/me to get the identity Jupyter assigned to the browser user.
Falls back to 'Anonymous' if unavailable.
"""
me = _api_get(f"{server_url}/api/me", token)
if me:
identity = me.get("identity", {})
return identity.get("display_name", "") or identity.get("username", "") or identity.get("name", "Anonymous")
return "Anonymous"
# ---------------------------------------------------------------------------
# Helpers nbformat
# ---------------------------------------------------------------------------
def _normalize_code_cell(cell: NotebookNode) -> dict:
"""Devuelve un dict de celda de codigo con todos los campos requeridos por nbformat.
Celdas creadas manualmente (no via Jupyter UI) pueden omitir 'outputs' o
'execution_count'. El modelo CRDT de jupyter_nbmodel_client accede a estos
campos sin comprobar su existencia, produciendo KeyError al ejecutar.
Este helper garantiza que el dict tenga la estructura completa.
"""
def _new_code_cell(source: str) -> dict:
"""Crea un dict de celda de codigo nbformat 4.5 con todos los campos."""
return {
"id": cell.get("id", ""),
"id": str(uuid.uuid4()),
"cell_type": "code",
"metadata": cell.get("metadata", {}),
"source": cell.get("source", ""),
"outputs": cell.get("outputs", []),
"execution_count": cell.get("execution_count", None),
"metadata": {},
"source": source,
"outputs": [],
"execution_count": None,
}
@@ -175,93 +172,18 @@ def _extract_outputs(raw_outputs: list[dict]) -> list[str]:
return result
# ---------------------------------------------------------------------------
# Modo append (async interno)
# ---------------------------------------------------------------------------
def _kernel_outputs_to_nbformat(outputs: list[dict]) -> list[dict]:
"""Normaliza outputs de KernelClient al esquema nbformat 4.
async def _async_append_execute(
notebook_path: str,
code: str,
server_url: str,
token: str,
) -> dict[str, Any]:
_create_notebook(notebook_path, server_url, token)
kernel_id = _ensure_session(server_url, token, notebook_path)
ws_url = get_jupyter_notebook_websocket_url(
server_url,
notebook_path,
token or None,
)
username = _resolve_collab_username(server_url, token)
async with NbModelClient(ws_url, username=username) as nb:
await nb.wait_until_synced()
with KernelClient(server_url=server_url, token=token, kernel_id=kernel_id) as kernel:
cell_index = nb.add_code_cell(code)
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None, partial(nb.execute_cell, cell_index, kernel),
)
# Let Y.js propagate changes to other clients (browser)
await asyncio.sleep(2)
outputs = _extract_outputs(result.get("outputs", []))
return {"cell_index": cell_index, "outputs": outputs}
KernelClient ya devuelve dicts con `output_type`, pero algunos casos (errores,
streams) pueden venir con campos sueltos. Esta funcion los pasa tal cual: el
cliente actual cumple el esquema; existe como punto de extension futuro.
"""
return [dict(o) for o in outputs]
# ---------------------------------------------------------------------------
# Modo cell (async interno)
# ---------------------------------------------------------------------------
async def _async_execute_cell(
notebook_path: str,
cell_index: int,
server_url: str,
token: str,
) -> dict[str, Any]:
kernel_id = _ensure_session(server_url, token, notebook_path)
ws_url = get_jupyter_notebook_websocket_url(
server_url,
notebook_path,
token or None,
)
username = _resolve_collab_username(server_url, token)
async with NbModelClient(ws_url, username=username) as nb:
await nb.wait_until_synced()
# Normalizar la celda antes de ejecutar. Las celdas creadas manualmente
# (sin pasar por la UI de Jupyter) pueden carecer de los campos 'outputs'
# o 'execution_count' en el modelo CRDT, lo que provoca KeyError dentro
# de execute_cell al intentar hacer `del ycell["outputs"][:]`.
# Reemplazar la celda via __setitem__ fuerza la re-creacion completa del
# mapa CRDT con todos los campos requeridos por nbformat.
cell = nb[cell_index]
if cell.get("cell_type") == "code" and (
"outputs" not in cell or "execution_count" not in cell
):
nb[cell_index] = _normalize_code_cell(cell)
with KernelClient(server_url=server_url, token=token, kernel_id=kernel_id) as kernel:
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None, partial(nb.execute_cell, cell_index, kernel),
)
await asyncio.sleep(2)
outputs = _extract_outputs(result.get("outputs", []))
return {"cell_index": cell_index, "outputs": outputs}
# ---------------------------------------------------------------------------
# API publica
# Modos
# ---------------------------------------------------------------------------
@@ -273,22 +195,31 @@ def jupyter_append_execute(
) -> dict[str, Any]:
"""Añade una celda de codigo al final del notebook y la ejecuta.
Tanto el agente como el usuario ven la celda y su output en tiempo real
porque la escritura se realiza a traves del protocolo colaborativo de Jupyter.
Args:
notebook_path: Ruta al notebook relativa a la raiz del servidor Jupyter.
code: Codigo Python a insertar y ejecutar.
server_url: URL del servidor Jupyter; por defecto http://localhost:8888.
token: Token de autenticacion del servidor Jupyter.
Returns:
dict con 'cell_index' (indice de la nueva celda) y 'outputs' (lista de strings).
Raises:
Exception: si no se puede conectar al servidor o al kernel.
Persiste la celda + outputs a disco via REST `/api/contents`. Jupyter Lab
detecta el cambio en el filesystem y lo refleja en el browser (puede pedir
'Revert to disk' segun version y conflictos).
"""
return asyncio.run(_async_append_execute(notebook_path, code, server_url, token))
_create_notebook(notebook_path, server_url, token)
kernel_id = _ensure_session(server_url, token, notebook_path)
# Lee notebook, añade celda nueva
file_node = _get_notebook_content(notebook_path, server_url, token)
nb = file_node["content"]
nb.setdefault("cells", [])
new_cell = _new_code_cell(code)
nb["cells"].append(new_cell)
cell_index = len(nb["cells"]) - 1
# Ejecuta en el kernel del notebook
with KernelClient(server_url=server_url, token=token, kernel_id=kernel_id) as kernel:
result = kernel.execute(code)
raw_outputs = result.get("outputs", [])
new_cell["outputs"] = _kernel_outputs_to_nbformat(raw_outputs)
new_cell["execution_count"] = result.get("execution_count")
_put_notebook_content(notebook_path, server_url, token, nb)
return {"cell_index": cell_index, "outputs": _extract_outputs(raw_outputs)}
def jupyter_execute_cell(
@@ -297,22 +228,32 @@ def jupyter_execute_cell(
server_url: str = "http://localhost:8888",
token: str = "",
) -> dict[str, Any]:
"""Ejecuta una celda existente del notebook por indice.
"""Ejecuta una celda existente por indice y persiste sus outputs."""
kernel_id = _ensure_session(server_url, token, notebook_path)
Args:
notebook_path: Ruta al notebook relativa a la raiz del servidor Jupyter.
cell_index: Indice de la celda a ejecutar (0-based).
server_url: URL del servidor Jupyter; por defecto http://localhost:8888.
token: Token de autenticacion del servidor Jupyter.
file_node = _get_notebook_content(notebook_path, server_url, token)
nb = file_node["content"]
cells = nb.get("cells", [])
if cell_index < 0 or cell_index >= len(cells):
raise IndexError(f"cell_index {cell_index} fuera de rango (notebook tiene {len(cells)} celdas)")
Returns:
dict con 'cell_index' y 'outputs' (lista de strings).
cell = cells[cell_index]
if cell.get("cell_type") != "code":
raise ValueError(f"La celda {cell_index} no es de codigo (cell_type={cell.get('cell_type')!r})")
Raises:
IndexError: si cell_index esta fuera de rango.
Exception: si no se puede conectar al servidor o al kernel.
"""
return asyncio.run(_async_execute_cell(notebook_path, cell_index, server_url, token))
source = cell.get("source", "")
if isinstance(source, list):
source = "".join(source)
with KernelClient(server_url=server_url, token=token, kernel_id=kernel_id) as kernel:
result = kernel.execute(source)
raw_outputs = result.get("outputs", [])
cell["outputs"] = _kernel_outputs_to_nbformat(raw_outputs)
cell["execution_count"] = result.get("execution_count")
_put_notebook_content(notebook_path, server_url, token, nb)
return {"cell_index": cell_index, "outputs": _extract_outputs(raw_outputs)}
def jupyter_kernel_execute(
@@ -320,24 +261,9 @@ def jupyter_kernel_execute(
server_url: str = "http://localhost:8888",
token: str = "",
) -> dict[str, Any]:
"""Ejecuta codigo directamente en el kernel sin modificar ningun notebook.
Util para consultas rapidas, inspeccion de variables, comprobaciones de estado.
Args:
code: Codigo Python a ejecutar en el kernel activo.
server_url: URL del servidor Jupyter; por defecto http://localhost:8888.
token: Token de autenticacion del servidor Jupyter.
Returns:
dict con 'outputs' (lista de strings) y 'status' ('ok' o 'error').
Raises:
Exception: si no se puede conectar al servidor o al kernel.
"""
"""Ejecuta codigo directo en el kernel sin tocar ningun notebook."""
with KernelClient(server_url=server_url, token=token) as kernel:
result = kernel.execute(code)
outputs = _extract_outputs(result.get("outputs", []))
return {"outputs": outputs, "status": result.get("status", "unknown")}
@@ -350,26 +276,21 @@ if __name__ == "__main__":
import argparse
import sys
parser = argparse.ArgumentParser(
description="Ejecuta codigo en kernels de Jupyter",
)
parser = argparse.ArgumentParser(description="Ejecuta codigo en kernels de Jupyter")
sub = parser.add_subparsers(dest="command", required=True)
# append
p_append = sub.add_parser("append", help="Añade celda al notebook y la ejecuta")
p_append.add_argument("notebook", help="Ruta al notebook relativa al servidor")
p_append.add_argument("code", help="Codigo a insertar y ejecutar")
p_append.add_argument("--server", default="http://localhost:8888")
p_append.add_argument("--token", default="")
# cell
p_cell = sub.add_parser("cell", help="Ejecuta celda existente por indice")
p_cell.add_argument("notebook", help="Ruta al notebook relativa al servidor")
p_cell.add_argument("index", type=int, help="Indice de la celda (0-based)")
p_cell.add_argument("--server", default="http://localhost:8888")
p_cell.add_argument("--token", default="")
# kernel
p_kernel = sub.add_parser("kernel", help="Ejecuta codigo en el kernel sin tocar notebook")
p_kernel.add_argument("code", help="Codigo a ejecutar")
p_kernel.add_argument("--server", default="http://localhost:8888")
python/functions/notebook/tests/__init__.py
View File
python/functions/notebook/tests/test_jupyter_exec.py
+188
View File
@@ -0,0 +1,188 @@
"""Tests para jupyter_exec.
Cubre:
- Que `_notebook_exists` usa GET (regresion del bug 0050: HEAD daba 405).
- Que `_create_notebook` no toca el servidor si el notebook ya existe.
- E2E contra un Jupyter Lab vivo si esta disponible (skip si no).
"""
from __future__ import annotations
import json
import os
import socket
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", ".."))
from python.functions.notebook import jupyter_exec as jx
# ---------------------------------------------------------------------------
# Tests unitarios (regresion del bug HEAD/GET)
# ---------------------------------------------------------------------------
def _http_response_mock(body: bytes = b"{}", status: int = 200) -> MagicMock:
resp = MagicMock()
resp.read.return_value = body
resp.__enter__ = lambda self: self
resp.__exit__ = lambda self, *a: False
resp.status = status
return resp
def test_notebook_exists_uses_get_not_head():
"""Regresion 0050: HEAD devuelve 405 en /api/contents; debe usar GET."""
captured = {}
def fake_urlopen(req, timeout):
captured["method"] = req.get_method()
captured["url"] = req.full_url
return _http_response_mock(b'{"name":"x.ipynb"}')
with patch.object(jx, "urlopen", side_effect=fake_urlopen):
ok = jx._notebook_exists("x.ipynb", "http://srv", "")
assert ok is True
assert captured["method"] == "GET"
assert "content=0" in captured["url"]
def test_notebook_exists_returns_false_on_404():
err = urllib.request.HTTPError(url="x", code=404, msg="nope", hdrs=None, fp=None)
with patch.object(jx, "urlopen", side_effect=err):
assert jx._notebook_exists("x.ipynb", "http://srv", "") is False
def test_create_notebook_skips_when_exists():
with patch.object(jx, "_notebook_exists", return_value=True), \
patch.object(jx, "urlopen") as mock_open:
jx._create_notebook("x.ipynb", "http://srv", "")
mock_open.assert_not_called()
def test_new_code_cell_has_required_fields():
cell = jx._new_code_cell("print(42)")
assert cell["cell_type"] == "code"
assert cell["source"] == "print(42)"
assert cell["outputs"] == []
assert cell["execution_count"] is None
assert isinstance(cell["id"], str) and len(cell["id"]) > 0
assert cell["metadata"] == {}
def test_extract_outputs_handles_streams_and_results():
raw = [
{"output_type": "stream", "name": "stdout", "text": "hola\n"},
{"output_type": "execute_result", "data": {"text/plain": "42"}},
{"output_type": "error", "traceback": ["E1", "E2"]},
]
out = jx._extract_outputs(raw)
assert out == ["hola", "42", "E1\nE2"]
# ---------------------------------------------------------------------------
# E2E (requiere Jupyter Lab corriendo)
# ---------------------------------------------------------------------------
JUPYTER_VENV_BIN = Path("/home/lucas/fn_registry/analysis/pruebas_jupyter/.venv/bin")
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _wait_http(url: str, timeout: float = 10.0) -> bool:
end = time.time() + timeout
while time.time() < end:
try:
with urllib.request.urlopen(url, timeout=1):
return True
except OSError:
time.sleep(0.3)
return False
@pytest.fixture(scope="module")
def jupyter_server(tmp_path_factory):
"""Arranca un Jupyter Lab en puerto libre. Skip si las deps no estan."""
if not (JUPYTER_VENV_BIN / "jupyter-lab").exists():
pytest.skip("Jupyter Lab no disponible en pruebas_jupyter venv")
workdir = tmp_path_factory.mktemp("jupyter_e2e")
(workdir / "notebooks").mkdir()
port = _free_port()
proc = subprocess.Popen(
[
str(JUPYTER_VENV_BIN / "jupyter-lab"),
f"--port={port}",
"--no-browser",
"--ServerApp.token=",
"--ServerApp.password=",
"--ServerApp.disable_check_xsrf=True",
f"--ServerApp.root_dir={workdir}",
"--collaborative",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
server_url = f"http://localhost:{port}"
if not _wait_http(f"{server_url}/api"):
proc.terminate()
pytest.skip("Jupyter Lab no levantó a tiempo")
yield server_url, workdir
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
def test_e2e_append_executes_and_persists(jupyter_server):
server_url, workdir = jupyter_server
result = jx.jupyter_append_execute(
"notebooks/test.ipynb", "z = 21 * 2; print(z)", server_url=server_url,
)
assert result["cell_index"] == 0
assert result["outputs"] == ["42"]
nb = json.loads((workdir / "notebooks" / "test.ipynb").read_text())
assert len(nb["cells"]) == 1
assert nb["cells"][0]["execution_count"] == 1
def test_e2e_append_twice_increments_index(jupyter_server):
server_url, _ = jupyter_server
jx.jupyter_append_execute("notebooks/twice.ipynb", "a = 1", server_url=server_url)
r2 = jx.jupyter_append_execute("notebooks/twice.ipynb", "print(a + 1)", server_url=server_url)
assert r2["cell_index"] == 1
assert r2["outputs"] == ["2"]
def test_e2e_cell_executes_existing(jupyter_server):
server_url, _ = jupyter_server
jx.jupyter_append_execute("notebooks/cell.ipynb", "v = 10", server_url=server_url)
jx.jupyter_append_execute("notebooks/cell.ipynb", "print(v * 5)", server_url=server_url)
r = jx.jupyter_execute_cell("notebooks/cell.ipynb", 1, server_url=server_url)
assert r["outputs"] == ["50"]
def test_e2e_kernel_mode(jupyter_server):
server_url, _ = jupyter_server
r = jx.jupyter_kernel_execute("print('hello kernel')", server_url=server_url)
assert r["status"] == "ok"
assert r["outputs"] == ["hello kernel"]