wehub-resource-sync
293 lines
9.2 KiB
Python
293 lines
9.2 KiB
Python
"""Tests for browser-opening helpers used by CLI frontends."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
import omnigent.conversation_browser as browser
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url,expected",
|
|
[
|
|
# Databricks workspace API mount → the recognizable /omnigent SPA URL.
|
|
(
|
|
"https://e2-dogfood.staging.cloud.databricks.com/api/2.0/omnigent",
|
|
"https://e2-dogfood.staging.cloud.databricks.com/omnigent",
|
|
),
|
|
# A trailing ``?o=<org>`` selector on the API base is dropped.
|
|
(
|
|
"https://ws.databricks.com/api/2.0/omnigent?o=123",
|
|
"https://ws.databricks.com/omnigent",
|
|
),
|
|
# Trailing slash on the API mount still maps cleanly.
|
|
(
|
|
"https://ws.databricks.com/api/2.0/omnigent/",
|
|
"https://ws.databricks.com/omnigent",
|
|
),
|
|
# Non-Databricks URLs pass through unchanged (sans trailing slash).
|
|
("http://127.0.0.1:6767", "http://127.0.0.1:6767"),
|
|
("https://omnigent-02m5.onrender.com/", "https://omnigent-02m5.onrender.com"),
|
|
],
|
|
)
|
|
def test_display_server_url_maps_databricks_api_mount(url: str, expected: str) -> None:
|
|
"""
|
|
``display_server_url`` rewrites the Databricks API mount to the SPA URL.
|
|
|
|
What this proves: the startup banner shows the workspace ``/omnigent``
|
|
URL a user recognizes instead of the internal ``/api/2.0/omnigent``
|
|
proxy path, while every other target is shown verbatim. A regression
|
|
that stopped mapping would leak the API path back into the banner.
|
|
|
|
:returns: None.
|
|
"""
|
|
assert browser.display_server_url(url) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url,expected",
|
|
[
|
|
("https://ws.databricks.com/api/2.0/omnigent", True),
|
|
("https://ws.databricks.com/api/2.0/omnigent/", True),
|
|
("https://ws.databricks.com/omnigent", False), # the SPA URL, not the API mount
|
|
("http://127.0.0.1:6767", False),
|
|
("https://omnigent-02m5.onrender.com", False),
|
|
],
|
|
)
|
|
def test_is_workspace_hosted_url(url: str, expected: bool) -> None:
|
|
"""
|
|
``is_workspace_hosted_url`` is true only for the workspace API mount.
|
|
|
|
What this proves: the predicate the banner uses to suppress the
|
|
server-version row fires for ``/api/2.0/omnigent`` and nothing else, so
|
|
non-Databricks targets keep showing their version.
|
|
|
|
:returns: None.
|
|
"""
|
|
assert browser.is_workspace_hosted_url(url) is expected
|
|
|
|
|
|
def test_conversation_url_quotes_session_id() -> None:
|
|
"""
|
|
Conversation URLs percent-encode ids before appending them to the base URL.
|
|
|
|
:returns: None.
|
|
"""
|
|
url = browser.conversation_url(
|
|
"https://example.com/app/",
|
|
"conv with/slash?query",
|
|
)
|
|
|
|
assert url == "https://example.com/app/c/conv%20with%2Fslash%3Fquery"
|
|
|
|
|
|
def test_open_conversation_url_uses_macos_open(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""
|
|
macOS launches use the platform ``open`` command with the URL as one argv.
|
|
|
|
:param monkeypatch: Pytest monkeypatch fixture.
|
|
:returns: None.
|
|
"""
|
|
calls: list[tuple[list[str], object, object, bool]] = []
|
|
|
|
def fake_run(
|
|
args: list[str],
|
|
*,
|
|
stdout: object,
|
|
stderr: object,
|
|
check: bool,
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
"""
|
|
Capture the subprocess launch request.
|
|
|
|
:param args: Command argv, e.g. ``["open", "http://..."]``.
|
|
:param stdout: Captured stdout target.
|
|
:param stderr: Captured stderr target.
|
|
:param check: Whether non-zero exits should raise.
|
|
:returns: Completed process object for the fake ``open`` command.
|
|
"""
|
|
calls.append((args, stdout, stderr, check))
|
|
return subprocess.CompletedProcess(args=args, returncode=0)
|
|
|
|
monkeypatch.setattr(browser.sys, "platform", "darwin")
|
|
monkeypatch.setattr(browser.subprocess, "run", fake_run)
|
|
|
|
opened = browser.open_conversation_url("http://127.0.0.1:8000/c/conv_abc")
|
|
|
|
assert opened is True
|
|
assert calls == [
|
|
(
|
|
["open", "http://127.0.0.1:8000/c/conv_abc"],
|
|
browser.subprocess.DEVNULL,
|
|
browser.subprocess.DEVNULL,
|
|
False,
|
|
)
|
|
]
|
|
|
|
|
|
def test_open_conversation_link_skips_when_disabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""
|
|
Disabled automatic browser opens return before building or opening a URL.
|
|
|
|
:param monkeypatch: Pytest monkeypatch fixture.
|
|
:returns: None.
|
|
"""
|
|
warnings: list[str] = []
|
|
|
|
def fail_open(url: str) -> bool:
|
|
"""
|
|
Fail if the disabled guard calls the opener.
|
|
|
|
:param url: Browser URL, e.g. ``"http://localhost/c/conv_abc"``.
|
|
:returns: Never returns; this stub always fails the test.
|
|
:raises AssertionError: Always raised when called.
|
|
"""
|
|
raise AssertionError(f"disabled guard should not open {url}")
|
|
|
|
monkeypatch.setattr(browser, "open_conversation_url", fail_open)
|
|
|
|
browser.open_conversation_link_if_enabled(
|
|
base_url="http://127.0.0.1:8000/",
|
|
conversation_id="conv abc",
|
|
enabled=False,
|
|
warn=warnings.append,
|
|
)
|
|
|
|
assert warnings == []
|
|
|
|
|
|
def test_open_conversation_link_warns_when_opener_declines(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""
|
|
Failed opener attempts surface a warning with the conversation URL.
|
|
|
|
:param monkeypatch: Pytest monkeypatch fixture.
|
|
:returns: None.
|
|
"""
|
|
warnings: list[str] = []
|
|
|
|
def fake_open(url: str) -> bool:
|
|
"""
|
|
Simulate a platform opener that declines the URL.
|
|
|
|
:param url: Browser URL, e.g. ``"http://localhost/c/conv_abc"``.
|
|
:returns: ``False`` to signal that no opener accepted the URL.
|
|
"""
|
|
return False
|
|
|
|
monkeypatch.setattr(browser, "open_conversation_url", fake_open)
|
|
|
|
browser.open_conversation_link_if_enabled(
|
|
base_url="http://127.0.0.1:8000/",
|
|
conversation_id="conv abc",
|
|
enabled=True,
|
|
warn=warnings.append,
|
|
)
|
|
|
|
assert warnings == [
|
|
"Warning: no browser opener accepted conversation URL http://127.0.0.1:8000/c/conv%20abc"
|
|
]
|
|
|
|
|
|
def test_open_conversation_link_warns_when_opener_raises_oserror(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""
|
|
OSError from the platform opener is surfaced through the warning callback.
|
|
|
|
:param monkeypatch: Pytest monkeypatch fixture.
|
|
:returns: None.
|
|
"""
|
|
warnings: list[str] = []
|
|
|
|
def fail_open(url: str) -> bool:
|
|
"""
|
|
Simulate an opener executable that cannot be started.
|
|
|
|
:param url: Browser URL, e.g. ``"http://localhost/c/conv_abc"``.
|
|
:returns: Never returns because the opener raises.
|
|
:raises OSError: Always raised to simulate a missing opener.
|
|
"""
|
|
raise OSError("missing opener")
|
|
|
|
monkeypatch.setattr(browser, "open_conversation_url", fail_open)
|
|
|
|
browser.open_conversation_link_if_enabled(
|
|
base_url="http://127.0.0.1:8000/",
|
|
conversation_id="conv abc",
|
|
enabled=True,
|
|
warn=warnings.append,
|
|
)
|
|
|
|
assert warnings == [
|
|
"Warning: failed to open conversation URL "
|
|
"http://127.0.0.1:8000/c/conv%20abc: missing opener"
|
|
]
|
|
|
|
|
|
def test_conversation_url_maps_workspace_hosted_server_to_ui_mount(tmp_path, monkeypatch) -> None:
|
|
"""Workspace-hosted servers link to the SPA mount with the org selector.
|
|
|
|
The server base is the API proxy (``/api/2.0/omnigent``) — linking
|
|
there returns JSON, not the web UI. The browser URL must land on
|
|
``/omnigent`` and carry ``?o=<org>`` recorded by ``omnigent
|
|
login`` so multi-org workspaces open in the right one.
|
|
"""
|
|
from omnigent.cli_auth import store_databricks_auth
|
|
from omnigent.conversation_browser import conversation_url
|
|
|
|
monkeypatch.setattr(
|
|
"omnigent.cli_auth._token_file_path",
|
|
lambda: tmp_path / "auth_tokens.json",
|
|
)
|
|
server = "https://example.databricks.com/api/2.0/omnigent"
|
|
store_databricks_auth(
|
|
server,
|
|
"https://example.databricks.com",
|
|
org_id="2850744067564480",
|
|
)
|
|
|
|
url = conversation_url(server, "conv_abc123")
|
|
|
|
assert url == ("https://example.databricks.com/omnigent/c/conv_abc123?o=2850744067564480")
|
|
|
|
|
|
def test_conversation_url_workspace_hosted_without_org_record(tmp_path, monkeypatch) -> None:
|
|
"""No recorded org id → SPA mount link without the ?o selector.
|
|
|
|
Single-org workspaces resolve fine without it; inventing an org id
|
|
would be worse than omitting it.
|
|
"""
|
|
from omnigent.conversation_browser import conversation_url
|
|
|
|
monkeypatch.setattr(
|
|
"omnigent.cli_auth._token_file_path",
|
|
lambda: tmp_path / "auth_tokens.json",
|
|
)
|
|
|
|
url = conversation_url("https://example.databricks.com/api/2.0/omnigent", "conv_abc123")
|
|
|
|
assert url == "https://example.databricks.com/omnigent/c/conv_abc123"
|
|
|
|
|
|
def test_conversation_url_plain_server_unchanged(tmp_path, monkeypatch) -> None:
|
|
"""Non-workspace servers keep the plain /c/<id> link shape."""
|
|
from omnigent.conversation_browser import conversation_url
|
|
|
|
monkeypatch.setattr(
|
|
"omnigent.cli_auth._token_file_path",
|
|
lambda: tmp_path / "auth_tokens.json",
|
|
)
|
|
|
|
assert (
|
|
conversation_url("http://127.0.0.1:6767", "conv_abc123")
|
|
== "http://127.0.0.1:6767/c/conv_abc123"
|
|
)
|