unsloth/studio/backend/tests/test_colab_embed.py
Souravrajvi0 d17567af3e
fix(studio/colab): restore blank Colab iframe embed (#7344) (#7349)
* fix(studio/colab): restore iframe embed via serve_kernel_port_as_iframe

Colab's output sanitizer often strips custom <iframe> tags from
IPython.display.HTML without raising, leaving a blank cell even though
display() succeeded. The kernel-port helper is the supported embedding
path and registers the proxy correctly.

- Prefer serve_kernel_port_as_iframe; keep raw HTML iframe as fallback
- Always show the clickable link card via show_link() so the proxy URL
  is visible even when iframe embedding fails
- Add regression tests for embed ordering and URL truncation

Fixes #7344

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(studio/colab): harden iframe embed fallbacks per Codex review

Guard show_link so a display failure cannot skip embedding, and only use
serve_kernel_port_as_iframe when get_colab_url returned a real Colab proxy
URL so localhost/colabtools environments still get the HTML iframe path.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(studio/colab): stop opening Colab proxy URLs in a new tab (#7349)

Colab *.prod.colab.dev proxy hosts are session-scoped and return HTTP 404
when opened as a top-level tab or from another device. Replace the
clickable Open button for those URLs with an in-notebook ready card, keep
serve_kernel_port_as_iframe for the UI, and point users at
start(cloudflare=True) for a real shareable / new-window link.

* fix(studio/colab): use kernel iframe on real Colab when eval_js fails (#7349)

Gate serve_kernel_port_as_iframe on COLAB_RELEASE_TAG + google.colab import
instead of a successful proxyPort URL. When eval_js fails and get_colab_url
falls back to localhost, real Colab notebooks still embed via the kernel helper
(port-only). colabtools without COLAB_RELEASE_TAG keeps the HTML iframe path.

Thanks @mfielding92 for the runtime diagnosis.

* Mock top-level google package in Colab embed tests

* test(studio/colab): mock top-level google package in Colab tests

Patching only sys.modules["google.colab"] fails when no google namespace
is installed: import google.colab resolves the parent first and returns
False in _is_colab_runtime(). Add a shared helper that mocks both google
and google.colab for deterministic tests across environments.

* Tighten comments in Colab embed helpers and tests

* fix(studio/colab): default Cloudflare on Colab with durable login credentials

Colab proxy iframes often load an empty document even when the kernel helper
appends the frame, leaving users unable to reach Studio to change the bootstrap
password and blocking start(cloudflare=True).

On real Colab runtime:
- Default cloudflare to True (pass cloudflare=False to opt out)
- Finalize the random admin password and print credentials in the notebook
- Persist credentials across cell re-runs after interrupt
- Show Cloudflare link before login credentials; skip blank proxy iframe when ready
- Reuse main._IS_COLAB for runtime detection (not COLAB_RELEASE_TAG alone)
- Only trust serve_kernel_port_as_iframe on real Colab; colabtools falls back to HTML
- Keep embedding when the link card display fails

Addresses Codex review feedback on #7349 and @mfielding92's catch-22 report.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(studio/colab): skip credential finalize when cloudflare=False

Only call _finalize_colab_admin_password() when opening a Cloudflare
tunnel. start(cloudflare=False) should not clear the bootstrap-password
gate or show a login card that references a missing tunnel link.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(studio/colab): drop stale cached Colab credentials after password change

On a Colab rerun the finalize path redisplayed the cached first-run
password whenever the bootstrap gate was already cleared. If the admin
changed the password through the app, that cached copy no longer
authenticates, so the notebook printed dead credentials. Validate the
cached password against the current stored hash before redisplaying and
drop the cache when it no longer matches.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Daniel Han <danielhanchen@gmail.com>
2026-07-24 02:23:24 -07:00

479 lines
17 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Regression coverage for Colab iframe embedding (#7344)."""
import sys
import types
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import colab
def _mock_google_colab_modules(colab_mod):
"""Mock ``google`` and ``google.colab`` for environments without Google packages."""
google_mod = types.ModuleType("google")
google_mod.colab = colab_mod
return {"google": google_mod, "google.colab": colab_mod}
def test_short_colab_url_truncates_proxy_host():
url = "https://8888-gpu-a100-s-kkb-usc1f0-9hzedjcxrlu8-f.us-central1-0.prod.colab.dev/"
assert colab._short_colab_url(url, 8888) == "https://8888-gpu-..."
def test_short_colab_url_falls_back_on_unexpected_shape():
assert colab._short_colab_url("https://example.com", 8888) == "https://example.com"
def test_is_colab_proxy_url_requires_https_proxy():
assert colab._is_colab_proxy_url("https://8888-test.prod.colab.dev/", 8888) is True
assert colab._is_colab_proxy_url("http://localhost:8888", 8888) is False
assert colab._is_colab_proxy_url("http://127.0.0.1:8888", 8888) is False
def test_ready_card_html_does_not_open_colab_proxy_in_new_tab():
"""Colab proxy hosts 404 as top-level tabs (#7349 reporter); never window.open them."""
html = colab._ready_card_html("https://8888-test.prod.colab.dev/", 8888)
assert "window.open" not in html
assert 'href="https://8888-test.prod.colab.dev/"' not in html
assert "start(cloudflare=True)" in html
def test_ready_card_html_points_to_cloudflare_when_link_ready(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
html = colab._ready_card_html(
"https://8888-test.prod.colab.dev/",
8888,
has_cloudflare_link = True,
)
assert "Cloudflare link above" in html
def test_ready_card_html_warns_when_cloudflare_tunnel_missing(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
html = colab._ready_card_html(
"https://8888-test.prod.colab.dev/",
8888,
cloudflare_requested = True,
)
assert "Could not open a Cloudflare tunnel" in html
def test_warn_colab_cloudflare_missing_logs_on_colab_without_tunnel(monkeypatch):
warnings: list[str] = []
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(colab.logger, "warning", lambda msg, **kwargs: warnings.append(msg))
colab._warn_colab_cloudflare_missing(use_cloudflare = True, cloudflare_url = None)
assert warnings
assert "Cloudflare tunnel unavailable" in warnings[0]
def test_warn_colab_cloudflare_missing_skips_when_tunnel_ready(monkeypatch, caplog):
import logging
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
with caplog.at_level(logging.WARNING):
colab._warn_colab_cloudflare_missing(
use_cloudflare = True,
cloudflare_url = "https://share.trycloudflare.com",
)
assert "Cloudflare tunnel unavailable" not in caplog.text
def test_is_colab_runtime_uses_backend_colab_detector(monkeypatch):
fake_main = types.ModuleType("main")
fake_main._IS_COLAB = True
monkeypatch.setitem(sys.modules, "main", fake_main)
assert colab._is_colab_runtime() is True
fake_main._IS_COLAB = False
assert colab._is_colab_runtime() is False
def test_ready_card_html_uses_cloudflare_hint_on_colab_runtime_localhost(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
html = colab._ready_card_html("http://localhost:8888", 8888)
assert "window.open" not in html
assert "start(cloudflare=True)" in html
def test_ready_card_html_keeps_open_button_for_localhost_outside_colab(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: False)
html = colab._ready_card_html("http://localhost:8888", 8888)
assert "window.open" in html
assert 'href="http://localhost:8888"' in html
assert "Open Unsloth Studio" in html
def test_embed_kernel_port_iframe_uses_colab_helper(monkeypatch):
colab_output = MagicMock()
google_colab = SimpleNamespace(output = colab_output)
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
with patch.dict("sys.modules", _mock_google_colab_modules(google_colab)):
assert colab._embed_kernel_port_iframe(8888) is True
colab_output.serve_kernel_port_as_iframe.assert_called_once_with(
8888,
height = colab._COLAB_IFRAME_HEIGHT,
width = "100%",
)
def test_embed_kernel_port_iframe_returns_false_without_colab():
with patch.dict("sys.modules", _mock_google_colab_modules(None)):
assert colab._embed_kernel_port_iframe(8888) is False
def test_embed_kernel_port_iframe_skips_colabtools_without_runtime(monkeypatch):
"""colabtools can queue JS without appending an iframe; only trust the helper on Colab."""
colab_output = MagicMock()
google_colab = SimpleNamespace(output = colab_output)
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: False)
with patch.dict("sys.modules", _mock_google_colab_modules(google_colab)):
assert colab._embed_kernel_port_iframe(8888) is False
colab_output.serve_kernel_port_as_iframe.assert_not_called()
def test_show_and_embed_prefers_kernel_port_iframe(monkeypatch):
calls: list[str] = []
monkeypatch.setattr(colab, "get_colab_url", lambda port: f"https://{port}-test.prod.colab.dev/")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"show_link",
lambda port,
*,
_url = None,
has_cloudflare_link = False,
cloudflare_requested = False: calls.append("show_link"),
)
monkeypatch.setattr(
colab,
"_embed_kernel_port_iframe",
lambda port: calls.append("kernel_iframe") or True,
)
monkeypatch.setattr(
colab,
"_embed_html_iframe",
lambda url, port: calls.append("html_iframe") or True,
)
colab._show_and_embed(8888)
assert calls == ["show_link", "kernel_iframe"]
def test_show_and_embed_falls_back_to_html_iframe(monkeypatch):
calls: list[str] = []
monkeypatch.setattr(colab, "get_colab_url", lambda port: f"https://{port}-test.prod.colab.dev/")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: False)
monkeypatch.setattr(
colab,
"show_link",
lambda port, *, _url = None, has_cloudflare_link = False: None,
)
monkeypatch.setattr(colab, "_embed_kernel_port_iframe", lambda port: False)
monkeypatch.setattr(
colab,
"_embed_html_iframe",
lambda url, port: calls.append((url, port)) or True,
)
colab._show_and_embed(8888)
assert calls == [("https://8888-test.prod.colab.dev/", 8888)]
def test_colab_wants_cloudflare_auto_enables_on_runtime(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
assert colab._colab_wants_cloudflare(None) is True
assert colab._colab_wants_cloudflare(True) is True
assert colab._colab_wants_cloudflare(False) is False
def test_colab_wants_cloudflare_defaults_off_outside_runtime(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: False)
assert colab._colab_wants_cloudflare(None) is False
assert colab._colab_wants_cloudflare(True) is True
def test_finalize_colab_admin_password_skips_outside_runtime(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: False)
assert colab._finalize_colab_admin_password() is None
def test_finalize_colab_admin_password_clears_bootstrap_gate(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(colab, "_load_colab_login_credentials", lambda: None)
stored: list[tuple[str, str]] = []
monkeypatch.setattr(
colab,
"_store_colab_login_credentials",
lambda username, password: stored.append((username, password)),
)
storage = SimpleNamespace(
DEFAULT_ADMIN_USERNAME = "unsloth",
ensure_default_admin = MagicMock(),
get_bootstrap_password = MagicMock(return_value = "alpha-beta-gamma"),
generate_bootstrap_password = MagicMock(return_value = "alpha-beta-gamma"),
requires_password_change = MagicMock(return_value = True),
update_password = MagicMock(return_value = True),
)
auth_pkg = types.ModuleType("auth")
auth_pkg.storage = storage
with patch.dict("sys.modules", {"auth": auth_pkg, "auth.storage": storage}):
result = colab._finalize_colab_admin_password()
assert result == ("unsloth", "alpha-beta-gamma")
storage.ensure_default_admin.assert_called_once()
storage.update_password.assert_called_once_with("unsloth", "alpha-beta-gamma")
assert stored == [("unsloth", "alpha-beta-gamma")]
def test_start_skips_finalize_when_cloudflare_disabled(monkeypatch):
import time
finalize_calls: list[str] = []
monkeypatch.setattr(colab, "_is_studio_healthy", lambda port: True)
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"_finalize_colab_admin_password",
lambda: finalize_calls.append("finalize") or ("unsloth", "secret"),
)
monkeypatch.setattr(
colab, "start_cloudflare_tunnel", lambda port: "https://share.trycloudflare.com"
)
monkeypatch.setattr(colab, "_publish_cloudflare_url", lambda url: None)
monkeypatch.setattr(colab, "_show_and_embed", lambda port, **kwargs: None)
monkeypatch.setattr(colab, "_stop_cloudflare_tunnel", lambda: None)
monkeypatch.setattr(time, "sleep", lambda _: (_ for _ in ()).throw(KeyboardInterrupt))
colab.start(cloudflare = False)
assert finalize_calls == []
def test_finalize_colab_admin_password_redisplay_on_rerun(monkeypatch):
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"_load_colab_login_credentials",
lambda: ("unsloth", "saved-pass"),
)
monkeypatch.setattr(colab, "_colab_credentials_still_valid", lambda username, password: True)
storage = SimpleNamespace(
DEFAULT_ADMIN_USERNAME = "unsloth",
ensure_default_admin = MagicMock(),
get_bootstrap_password = MagicMock(),
generate_bootstrap_password = MagicMock(),
requires_password_change = MagicMock(return_value = False),
update_password = MagicMock(),
)
auth_pkg = types.ModuleType("auth")
auth_pkg.storage = storage
with patch.dict("sys.modules", {"auth": auth_pkg, "auth.storage": storage}):
result = colab._finalize_colab_admin_password()
assert result == ("unsloth", "saved-pass")
storage.update_password.assert_not_called()
def test_finalize_colab_admin_password_drops_stale_cached_credentials(monkeypatch):
"""After an in-app password change the cached first-run password no longer
authenticates, so it must not be redisplayed (#7349 Codex review)."""
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"_load_colab_login_credentials",
lambda: ("unsloth", "stale-pass"),
)
monkeypatch.setattr(colab, "_colab_credentials_still_valid", lambda username, password: False)
cleared: list[bool] = []
monkeypatch.setattr(colab, "_clear_colab_login_credentials", lambda: cleared.append(True))
storage = SimpleNamespace(
DEFAULT_ADMIN_USERNAME = "unsloth",
ensure_default_admin = MagicMock(),
get_bootstrap_password = MagicMock(),
generate_bootstrap_password = MagicMock(),
requires_password_change = MagicMock(return_value = False),
update_password = MagicMock(),
)
auth_pkg = types.ModuleType("auth")
auth_pkg.storage = storage
with patch.dict("sys.modules", {"auth": auth_pkg, "auth.storage": storage}):
result = colab._finalize_colab_admin_password()
assert result is None
assert cleared == [True]
storage.update_password.assert_not_called()
def test_colab_credentials_still_valid_matches_stored_hash(monkeypatch):
from auth.hashing import hash_password
salt, pwd_hash = hash_password("right-pass")
storage = SimpleNamespace(
get_user_and_secret = MagicMock(return_value = (salt, pwd_hash, "jwt", False)),
)
with patch.dict("sys.modules", {"auth.storage": storage}):
assert colab._colab_credentials_still_valid("unsloth", "right-pass") is True
assert colab._colab_credentials_still_valid("unsloth", "wrong-pass") is False
def test_colab_credentials_still_valid_false_when_user_missing(monkeypatch):
storage = SimpleNamespace(get_user_and_secret = MagicMock(return_value = None))
with patch.dict("sys.modules", {"auth.storage": storage}):
assert colab._colab_credentials_still_valid("unsloth", "any") is False
def test_colab_login_html_includes_credentials():
html = colab._colab_login_html("unsloth", "alpha-beta-gamma-delta")
assert "unsloth" in html
assert "alpha-beta-gamma-delta" in html
def test_show_and_embed_renders_cloudflare_before_colab_login(monkeypatch):
displayed: list[str] = []
ipython_display = SimpleNamespace(
HTML = lambda html: SimpleNamespace(html = html),
display = lambda html: displayed.append(html.html),
)
monkeypatch.setattr(colab, "get_colab_url", lambda port: "https://8888-test.prod.colab.dev/")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"show_link",
lambda port, *, _url = None, has_cloudflare_link = False, cloudflare_requested = False: None,
)
monkeypatch.setattr(colab, "_embed_kernel_port_iframe", lambda port: True)
with patch.dict("sys.modules", {"IPython.display": ipython_display}):
colab._show_and_embed(
8888,
cloudflare_url = "https://share.trycloudflare.com",
colab_login = ("unsloth", "secret-pass"),
)
assert len(displayed) == 2
assert "share.trycloudflare.com" in displayed[0]
assert "secret-pass" in displayed[1]
def test_show_and_embed_skips_iframe_on_colab_when_cloudflare_ready(monkeypatch):
calls: list[str] = []
monkeypatch.setattr(colab, "get_colab_url", lambda port: f"https://{port}-test.prod.colab.dev/")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"show_link",
lambda port, *, _url = None, has_cloudflare_link = False, cloudflare_requested = False: None,
)
monkeypatch.setattr(
colab,
"_embed_kernel_port_iframe",
lambda port: calls.append("kernel_iframe") or True,
)
monkeypatch.setattr(
colab,
"_embed_html_iframe",
lambda url, port: calls.append("html_iframe") or True,
)
colab._show_and_embed(8888, cloudflare_url = "https://share.trycloudflare.com")
assert calls == []
def test_show_and_embed_uses_kernel_helper_on_colab_runtime_despite_localhost(monkeypatch):
calls: list[str] = []
monkeypatch.setattr(colab, "get_colab_url", lambda port: f"http://localhost:{port}")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"show_link",
lambda port,
*,
_url = None,
has_cloudflare_link = False,
cloudflare_requested = False: calls.append("show_link"),
)
monkeypatch.setattr(
colab,
"_embed_kernel_port_iframe",
lambda port: calls.append("kernel_iframe") or True,
)
monkeypatch.setattr(
colab,
"_embed_html_iframe",
lambda url, port: calls.append("html_iframe") or True,
)
colab._show_and_embed(8888)
assert calls == ["show_link", "kernel_iframe"]
def test_show_and_embed_skips_kernel_helper_for_localhost_outside_colab(monkeypatch):
calls: list[str] = []
monkeypatch.setattr(colab, "get_colab_url", lambda port: f"http://localhost:{port}")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: False)
monkeypatch.setattr(
colab,
"show_link",
lambda port,
*,
_url = None,
has_cloudflare_link = False,
cloudflare_requested = False: calls.append("show_link"),
)
monkeypatch.setattr(
colab,
"_embed_kernel_port_iframe",
lambda port: calls.append("kernel_iframe") or True,
)
monkeypatch.setattr(
colab,
"_embed_html_iframe",
lambda url, port: calls.append("html_iframe") or True,
)
colab._show_and_embed(8888)
assert calls == ["show_link", "html_iframe"]
def test_show_and_embed_still_embeds_when_show_link_fails(monkeypatch):
calls: list[str] = []
monkeypatch.setattr(colab, "get_colab_url", lambda port: f"https://{port}-test.prod.colab.dev/")
monkeypatch.setattr(colab, "_is_colab_runtime", lambda: True)
monkeypatch.setattr(
colab,
"show_link",
lambda port, *, _url = None: (_ for _ in ()).throw(RuntimeError("no display")),
)
monkeypatch.setattr(
colab,
"_embed_kernel_port_iframe",
lambda port: calls.append("kernel_iframe") or True,
)
monkeypatch.setattr(
colab,
"_embed_html_iframe",
lambda url, port: calls.append("html_iframe") or True,
)
colab._show_and_embed(8888)
assert calls == ["kernel_iframe"]