fn_registry/python/functions/notebook/tests/test_jupyter_exec.py

"""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(__file__).resolve().parents[4] / "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"]