feat: cierra issues 0050 y 0052 + commands automáticos
- 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:
@@ -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"]
|
||||
Reference in New Issue
Block a user