← Files Dataiku HeadlessARCHIVED FILE
tests/test_plugins.py
69.2 KB · Nov 11, 2026 · 12:58 UTC
# Copyright 2026 Dataiku SAS
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Unit tests for the instance plugin tools.
No live Dataiku instance is involved: ``get_dss_client`` is replaced with the fakes
below. Three Dataiku behaviours drive most of these cases and are asserted explicitly —
a failed plugin action is answered with HTTP 200 and ``success: false``; a code-env
action reports through a nested ``messages`` block whose own ``success`` field is false
even on success; and some actions complete inline with no future id at all.
"""
import asyncio
import io
import json
import zipfile
from unittest.mock import patch
import pytest
from dataiku_mcp.config import request
from dataiku_mcp.tools import plugins as tools
from tests.utils.fakes import FakeContext, incrementing_monotonic
_MANIFEST = "plugin.json"
_UNRESOLVED = (
"com.dataiku.dip.exceptions.CodedException: Types were missing when looking for "
"usages of the plugin geocoder; its deletion may be forced"
)
def _load(coro):
return json.loads(asyncio.run(coro))
def _listing_entry(plugin_id, version="1.0.0", dev=False, **meta):
return {
"id": plugin_id,
"version": version,
"isDev": dev,
"meta": {"label": plugin_id.replace("-", " ").title(), **meta},
}
def _alive():
return {"alive": True, "hasResult": False}
def _done(success=True, **result):
"""An update/delete future result, which has a top-level success flag."""
return {
"alive": False,
"aborted": False,
"hasResult": True,
"result": {"success": success, **result},
}
def _code_env_done(env_name, error=None):
"""A code-environment future result: no top-level success flag, ever.
Dataiku nests the outcome under ``messages``, and the success payload below is
verbatim from a real successful creation — ``messages.success`` is false even
though nothing failed, which is why only ``error``/``fatal`` may be trusted.
"""
if error is None:
messages = {
"maxSeverity": "INFO",
"anyMessage": True,
"success": False,
"warning": False,
"error": False,
"fatal": False,
"messages": [
{
"severity": "INFO",
"isFatal": False,
"code": "INFO_CODEENV_IMPORT_OK",
"message": "Import succeeded",
}
],
}
else:
messages = {
"maxSeverity": "ERROR",
"anyMessage": True,
"success": False,
"warning": False,
"error": True,
"fatal": True,
"messages": [
{
"severity": "ERROR",
"isFatal": True,
"code": "ERR_CODEENV_UPDATE_FAILED",
"message": error,
}
],
}
return {
"alive": False,
"aborted": False,
"hasResult": True,
"result": {"envName": env_name, "messages": messages},
}
class FakeFuture:
"""A DSSFuture stand-in walking a scripted sequence of states."""
def __init__(self, job_id, states):
self.job_id = job_id
self.state = states[0]
self._pending = list(states[1:])
def get_state(self):
if self._pending:
self.state = self._pending.pop(0)
self.read = True
return self.state
def peek_state(self):
return self.peeked
def _build_future(job_id, env_name, states):
"""A plugin code-env build, whose peeked state names its target environment."""
future = FakeFuture(job_id, states)
future.peeked = {
"jobId": job_id,
"payload": {"action": "import_env", "displayName": f"Import env: {env_name}"},
}
return future
class FakePluginSettings:
def __init__(self, raw):
self.raw = raw
def get_raw(self):
return self.raw
def set_code_env(self, name):
self.raw["codeEnvName"] = name
def save(self):
self.saves = getattr(self, "saves", 0) + 1
class FakeUsages:
def __init__(self, raw):
self.raw = raw
def get_raw(self):
return self.raw
class FakePlugin:
def __init__(self, plugin_id, settings_raw=None, usages_raw=None):
self.plugin_id = plugin_id
self.settings = FakePluginSettings(settings_raw or {})
self.usages_raw = usages_raw or {"usages": [], "missingTypes": []}
self.usages_error = None
self.delete_error = None
self.futures = {}
self.uploaded = None
self.delete_calls = []
# Set by FakeClient so an update makes the new version visible in the listing.
self.on_change = lambda: None
def get_settings(self):
return self.settings
def list_usages(self, project_key=None):
if self.usages_error is not None:
raise self.usages_error
return FakeUsages(self.usages_raw)
def update_from_store(self):
self.on_change()
return self.futures["update"]
def update_from_git(self, repository_url, checkout, subpath):
self.git_update_args = (repository_url, checkout, subpath)
self.on_change()
return self.futures["update"]
def start_update_from_zip(self, fp):
self.uploaded = fp.read()
self.on_change()
return self.futures["update"]
def create_code_env(self, python_interpreter=None):
self.created_interpreter = python_interpreter
if getattr(self, "create_error", None):
raise self.create_error
return self.futures["create_code_env"]
def update_code_env(self):
return self.futures["update_code_env"]
def delete(self, force=False):
self.delete_calls.append(force)
if self.delete_error is not None and not force:
raise self.delete_error
return self.futures["delete"]
class FakeClient:
host = "https://synthetic.example.com"
def __init__(
self, listing=None, plugins=None, listing_after_change=None, code_envs=None
):
self.code_envs = code_envs or []
self.futures = {}
self.listing = listing or []
self.plugins = plugins or {}
self.listing_after_change = listing_after_change
for plugin in self.plugins.values():
plugin.on_change = self._apply_change
def _apply_change(self):
"""Model Dataiku: after a successful action the listing reflects it."""
if self.listing_after_change is not None:
self.listing = self.listing_after_change
def list_plugins(self):
return [dict(entry) for entry in self.listing]
def get_plugin(self, plugin_id):
return self.plugins[plugin_id]
def list_code_envs(self):
return self.code_envs
def install_plugin_from_store(self, plugin_id):
self.store_install_id = plugin_id
self._apply_change()
return self.futures["install"]
def install_plugin_from_git(self, repository_url, checkout, subpath):
self.git_install_args = (repository_url, checkout, subpath)
self._apply_change()
return self.futures["install"]
def start_install_plugin_from_archive(self, archive):
self.uploaded = archive.read()
self._apply_change()
return self.futures["install"]
def _absent(plugin_id="geocoder", futures=None, version="1.0.0", **kwargs):
plugin = FakePlugin(plugin_id)
client = FakeClient(
plugins={plugin_id: plugin},
listing_after_change=[_listing_entry(plugin_id, version)],
)
client.futures.update(futures or {})
client.futures.update(kwargs)
return plugin, client
def _installed(plugin_id="geocoder", version="1.0.0", **kwargs):
"""A client where ``plugin_id`` is already installed."""
plugin = FakePlugin(
plugin_id,
settings_raw=kwargs.pop("settings", None),
usages_raw=kwargs.pop("usages", None),
)
plugin.futures.update(kwargs.pop("futures", {}))
client = FakeClient(
listing=[_listing_entry(plugin_id, version)],
plugins={plugin_id: plugin},
**kwargs,
)
return plugin, client
def _no_client(*args, **kwargs):
raise AssertionError(
"This test reached the real get_dss_client. Wrap it in _patch_client: the "
"suite must not depend on a configured Dataiku instance."
)
@pytest.fixture(autouse=True)
def _isolated(monkeypatch):
"""No real sleeping, and no route to a live instance.
Without the client guard a test can pass on a developer machine purely because
``~/.dataiku/config.json`` exists, and fail in CI where it does not.
"""
monkeypatch.setattr(tools, "_MIN_POLL_INTERVAL_SECONDS", 0)
monkeypatch.setattr(tools, "_MAX_POLL_INTERVAL_SECONDS", 0)
monkeypatch.setattr(tools, "get_dss_client", _no_client)
tools._BUILD_OWNERS.clear()
def _patch_client(client):
return patch("dataiku_mcp.tools.plugins.get_dss_client", return_value=client)
def _exhausted_clock():
return patch("dataiku_mcp.tools.plugins.time")
def _plugin_directory(tmp_path, plugin_id="my-plugin"):
directory = tmp_path / plugin_id
(directory / "python-lib").mkdir(parents=True)
(directory / _MANIFEST).write_text(
json.dumps({"id": plugin_id, "version": "0.1.0"})
)
(directory / "python-lib" / "core.py").write_text("VALUE = 1\n")
(directory / ".git").mkdir()
(directory / ".git" / "config").write_text("[core]\n")
return directory
def _members(payload):
return sorted(zipfile.ZipFile(io.BytesIO(payload)).namelist())
def test_update_plugin_local_path_is_unavailable_in_http_mode():
token = request.bind_http_identity("https://idp.example", "alice")
try:
with pytest.raises(ValueError, match="unavailable in HTTP mode"):
asyncio.run(
tools.update_plugin(
"local_path", FakeContext(), local_path="/server/plugin.zip"
)
)
finally:
request.reset_http_identity(token)
# --------------------------------------------------------------------------- #
# list_plugins
# --------------------------------------------------------------------------- #
def test_list_plugins_returns_catalog_metadata_with_summary_columns():
client = FakeClient(
listing=[
_listing_entry("geocoder", version="1.3.1", tags=["Geo"]),
_listing_entry("api-connect", version="1.2.1", deprecated=True),
]
)
with _patch_client(client):
res = _load(tools.list_plugins(FakeContext()))
assert res["total_plugins"] == 2
assert res["plugins"]["columns"] == ["id", "version", "dev", "label", "deprecated"]
assert res["plugins"]["rows"] == [
["api-connect", "1.2.1", False, "Api Connect", True],
["geocoder", "1.3.1", False, "Geocoder", False],
]
def test_list_plugins_partial_search_matches_label_and_tags_not_only_id():
client = FakeClient(
listing=[
_listing_entry("document-question-answering", label="Answers"),
_listing_entry("geocoder", tags=["Geography", "Enrichment"]),
_listing_entry("salesforce"),
]
)
with _patch_client(client):
by_label = _load(tools.list_plugins(FakeContext(), search="answers"))
by_tag = _load(tools.list_plugins(FakeContext(), search="geography"))
assert [row[0] for row in by_label["plugins"]["rows"]] == [
"document-question-answering"
]
assert [row[0] for row in by_tag["plugins"]["rows"]] == ["geocoder"]
def test_list_plugins_exact_search_returns_the_single_plugin():
client = FakeClient(
listing=[_listing_entry("geocoder"), _listing_entry("geocoder-extras")]
)
with _patch_client(client):
res = _load(
tools.list_plugins(FakeContext(), search="geocoder", search_mode="exact")
)
assert res["matched_plugins"] == 1
assert res["plugins"]["rows"][0][0] == "geocoder"
def test_list_plugins_filters_dev_and_deprecated_and_pages():
client = FakeClient(
listing=[
_listing_entry("a-dev", dev=True),
_listing_entry("b-old", deprecated=True),
_listing_entry("c-current"),
_listing_entry("d-current"),
]
)
with _patch_client(client):
dev_only = _load(tools.list_plugins(FakeContext(), dev=True))
current = _load(tools.list_plugins(FakeContext(), deprecated=False, limit=2))
page_two = _load(
tools.list_plugins(FakeContext(), deprecated=False, limit=2, offset=2)
)
assert [row[0] for row in dev_only["plugins"]["rows"]] == ["a-dev"]
assert [row[0] for row in current["plugins"]["rows"]] == ["a-dev", "c-current"]
assert current["next_offset"] == 2
assert [row[0] for row in page_two["plugins"]["rows"]] == ["d-current"]
assert page_two["next_offset"] is None
def test_list_plugins_details_report_code_env_but_never_plugin_config():
plugin = FakePlugin(
"api-connect",
settings_raw={
"codeEnvName": "plugin_api-connect_managed",
"config": {"api_key": "s3cret"},
"presets": [{"name": "prod", "config": {"token": "s3cret"}}],
},
)
client = FakeClient(
listing=[
_listing_entry(
"api-connect",
description="Call REST APIs",
author="Dataiku",
supportLevel="TIER2_SUPPORT",
licenseInfo="Apache Software License",
)
],
plugins={"api-connect": plugin},
)
with _patch_client(client):
res = _load(tools.list_plugins(FakeContext(), include_details=True))
row = dict(zip(res["plugins"]["columns"], res["plugins"]["rows"][0]))
assert row["code_env_name"] == "plugin_api-connect_managed"
assert row["support_level"] == "TIER2_SUPPORT"
assert "s3cret" not in json.dumps(res)
assert "presets" not in json.dumps(res)
def test_list_plugins_reads_settings_only_for_the_returned_page():
"""The free meta columns come from the listing; only code_env_name costs a call."""
reads = []
class CountingPlugin(FakePlugin):
def get_settings(self):
reads.append(self.plugin_id)
return self.settings
plugins = {f"p{i}": CountingPlugin(f"p{i}") for i in range(5)}
client = FakeClient(
listing=[_listing_entry(f"p{i}") for i in range(5)], plugins=plugins
)
with _patch_client(client):
_load(tools.list_plugins(FakeContext(), include_details=True, limit=2))
assert reads == ["p0", "p1"]
# --------------------------------------------------------------------------- #
# list_plugin_usages
# --------------------------------------------------------------------------- #
def test_list_plugin_usages_pages_all_deletion_blockers():
_, client = _installed(
usages={
"usages": [
{
"projectKey": "PROJ",
"objectType": "RECIPE",
"objectId": "geocode",
"extraField": "preserved",
}
],
"missingTypes": [
{"pluginId": "other", "missingType": "OtherType"},
{"pluginId": "geocoder", "missingType": "GeocodeType"},
],
}
)
with _patch_client(client):
first_page = _load(tools.list_plugin_usages("geocoder", FakeContext(), limit=1))
second_page = _load(
tools.list_plugin_usages("geocoder", FakeContext(), offset=1, limit=1)
)
assert first_page["blocker_count"] == 2
assert first_page["returned_blockers"] == 1
assert first_page["next_offset"] == 1
first_blocker = dict(
zip(first_page["blockers"]["columns"], first_page["blockers"]["rows"][0])
)
assert first_blocker["kind"] == "usage"
assert first_blocker["extraField"] == "preserved"
assert second_page["next_offset"] is None
second_blocker = dict(
zip(second_page["blockers"]["columns"], second_page["blockers"]["rows"][0])
)
assert second_blocker["kind"] == "unresolvable_component"
assert second_blocker["missingType"] == "GeocodeType"
@pytest.mark.parametrize(("offset", "limit"), [(-1, 1), (0, 0)])
def test_list_plugin_usages_validates_pagination_bounds(offset, limit):
with pytest.raises(ValueError, match="offset|limit"):
asyncio.run(
tools.list_plugin_usages(
"geocoder", FakeContext(), offset=offset, limit=limit
)
)
# --------------------------------------------------------------------------- #
# update_plugin
# --------------------------------------------------------------------------- #
def test_update_plugin_from_store_reports_completion_and_restart_need():
_, client = _installed(
version="1.3.1",
futures={
"update": FakeFuture(
"F1", [_alive(), _done(needsRestart=True, needsReload=True)]
)
},
)
with _patch_client(client):
res = _load(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
assert res["status"] == "completed"
assert res["operation"] == "update"
assert res["source"] == "store"
assert res["needs_restart"] is True
assert res["needs_reload"] is True
assert res["plugin"]["version"] == "1.3.1"
def test_update_plugin_reports_failure_reported_as_a_successful_response():
"""Dataiku answers a failed store update with HTTP 200 and success=false."""
_, client = _installed(
plugin_id="nope",
futures={
"update": FakeFuture(
None,
[
_done(
success=False,
installationError={
"message": "Plugin has since been removed from the store.",
"stackTraceStr": "com.dataiku.dip.exceptions.CodedException: …",
},
)
],
)
},
)
with _patch_client(client), pytest.raises(RuntimeError) as excinfo:
asyncio.run(tools.update_plugin("store", FakeContext(), plugin_id="nope"))
message = str(excinfo.value)
assert "removed from the store" in message
assert "CodedException" not in message
def test_update_plugin_from_local_path_reports_archive_failure(tmp_path):
directory = _plugin_directory(tmp_path)
_, client = _installed(
plugin_id="my-plugin",
futures={
"update": FakeFuture(
None,
[_done(success=False, installationError={"message": "bad archive"})],
)
},
)
with _patch_client(client), pytest.raises(RuntimeError) as excinfo:
asyncio.run(
tools.update_plugin("local_path", FakeContext(), local_path=str(directory))
)
assert "bad archive" in str(excinfo.value)
def test_update_plugin_completes_inline_when_dataiku_returns_no_future_id():
_, client = _installed(futures={"update": FakeFuture(None, [_done()])})
with _patch_client(client):
res = _load(
tools.update_plugin(
"store", FakeContext(), plugin_id="geocoder", wait_for_completion=False
)
)
assert res["status"] == "completed"
def test_update_plugin_without_waiting_returns_a_followable_future():
_, client = _installed(futures={"update": FakeFuture("F9", [_alive()])})
with _patch_client(client):
res = _load(
tools.update_plugin(
"store", FakeContext(), plugin_id="geocoder", wait_for_completion=False
)
)
assert res["status"] == "started"
assert res["operation"] == "update"
assert res["future_id"] == "F9"
assert "get_future_status" in res["hint"]
def test_update_plugin_timeout_reports_still_running_without_the_raw_state():
_, client = _installed(futures={"update": FakeFuture("F5", [_alive(), _alive()])})
with _patch_client(client), _exhausted_clock() as mock_time:
mock_time.monotonic.side_effect = incrementing_monotonic()
res = _load(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
assert res["status"] == "still_running"
assert res["future_id"] == "F5"
assert "state" not in res
assert "Do not start a duplicate operation" in res["hint"]
def test_update_plugin_rejects_arguments_that_do_not_match_the_source():
with pytest.raises(ValueError, match="local_path"):
asyncio.run(
tools.update_plugin(
"store", FakeContext(), plugin_id="x", local_path="/tmp/x"
)
)
with pytest.raises(ValueError, match="plugin_id"):
asyncio.run(tools.update_plugin("store", FakeContext()))
with pytest.raises(ValueError, match="local_path"):
asyncio.run(tools.update_plugin("local_path", FakeContext()))
def test_update_plugin_zips_a_directory_and_omits_version_control_files(tmp_path):
directory = _plugin_directory(tmp_path)
_, client = _installed(
plugin_id="my-plugin", futures={"update": FakeFuture(None, [_done()])}
)
with _patch_client(client):
res = _load(
tools.update_plugin("local_path", FakeContext(), local_path=str(directory))
)
assert res["status"] == "completed"
assert res["source"] == "local_path"
assert _members(client.plugins["my-plugin"].uploaded) == [
_MANIFEST,
"python-lib/core.py",
]
def test_update_plugin_rejects_a_symlinked_manifest(tmp_path):
directory = _plugin_directory(tmp_path)
manifest = directory / _MANIFEST
target = tmp_path / "manifest.json"
target.write_text(manifest.read_text())
manifest.unlink()
manifest.symlink_to(target)
with pytest.raises(ValueError, match="manifest.*symlink"):
asyncio.run(
tools.update_plugin("local_path", FakeContext(), local_path=str(directory))
)
def test_update_plugin_rejects_a_symlinked_file_in_a_directory(tmp_path):
directory = _plugin_directory(tmp_path)
secret = tmp_path / "secret.txt"
secret.write_text("do not upload")
(directory / "python-lib" / "secret.txt").symlink_to(secret)
with pytest.raises(ValueError, match="contains symlink"):
asyncio.run(
tools.update_plugin("local_path", FakeContext(), local_path=str(directory))
)
def test_update_plugin_flattens_a_zip_wrapped_in_one_directory(tmp_path):
archive_path = tmp_path / "download.zip"
with zipfile.ZipFile(archive_path, "w") as archive:
archive.writestr(
"dss-plugin-my-plugin/plugin.json", json.dumps({"id": "my-plugin"})
)
archive.writestr("dss-plugin-my-plugin/python-lib/core.py", "VALUE = 1\n")
_, client = _installed(
plugin_id="my-plugin", futures={"update": FakeFuture(None, [_done()])}
)
with _patch_client(client):
res = _load(
tools.update_plugin(
"local_path", FakeContext(), local_path=str(archive_path)
)
)
assert res["status"] == "completed"
assert _members(client.plugins["my-plugin"].uploaded) == [
_MANIFEST,
"python-lib/core.py",
]
def test_update_plugin_rejects_a_directory_without_a_manifest(tmp_path):
(tmp_path / "empty").mkdir()
with pytest.raises(ValueError, match="plugin.json"):
asyncio.run(
tools.update_plugin(
"local_path", FakeContext(), local_path=str(tmp_path / "empty")
)
)
def test_update_plugin_rejects_a_manifest_that_is_not_a_json_object(tmp_path):
directory = tmp_path / "broken"
directory.mkdir()
(directory / _MANIFEST).write_text('["not", "an", "object"]')
with pytest.raises(ValueError, match="must be a JSON object"):
asyncio.run(
tools.update_plugin("local_path", FakeContext(), local_path=str(directory))
)
def test_update_plugin_rejects_a_plugin_id_that_contradicts_the_manifest(tmp_path):
directory = _plugin_directory(tmp_path)
with _patch_client(FakeClient()), pytest.raises(ValueError) as excinfo:
asyncio.run(
tools.update_plugin(
"local_path",
FakeContext(),
plugin_id="other-plugin",
local_path=str(directory),
)
)
assert "my-plugin" in str(excinfo.value)
def test_update_plugin_rejects_a_missing_local_path():
with pytest.raises(ValueError, match="does not exist"):
asyncio.run(
tools.update_plugin(
"local_path", FakeContext(), local_path="/nonexistent/plugin"
)
)
def test_update_plugin_from_store_reports_the_resulting_version():
_, client = _installed(
version="1.3.1",
futures={"update": FakeFuture("F2", [_alive(), _done()])},
listing_after_change=[_listing_entry("geocoder", version="1.4.0")],
)
with _patch_client(client):
res = _load(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
assert res["status"] == "completed"
assert res["operation"] == "update"
assert res["plugin"]["version"] == "1.4.0"
def test_update_plugin_refuses_a_plugin_that_is_not_installed():
client = FakeClient(listing=[_listing_entry("geocoder-extras")])
with _patch_client(client), pytest.raises(ValueError) as excinfo:
asyncio.run(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
message = str(excinfo.value)
assert "not installed" in message
assert "geocoder-extras" in message
def test_update_plugin_rebuilds_the_bound_code_env():
_, client = _installed(
settings={"codeEnvName": "plugin_geocoder"},
futures={
"update": FakeFuture("F2", [_done()]),
"update_code_env": FakeFuture(
"F3", [_alive(), _code_env_done("plugin_geocoder")]
),
},
)
with _patch_client(client):
res = _load(
tools.update_plugin(
"store", FakeContext(), plugin_id="geocoder", rebuild_code_env=True
)
)
assert res["status"] == "completed"
assert res["code_env_rebuild"] == {
"status": "completed",
"code_env_name": "plugin_geocoder",
}
def test_update_plugin_reports_an_unbound_code_env_instead_of_failing():
_, client = _installed(futures={"update": FakeFuture("F2", [_done()])})
with _patch_client(client):
res = _load(
tools.update_plugin(
"store", FakeContext(), plugin_id="geocoder", rebuild_code_env=True
)
)
assert res["status"] == "completed"
assert res["code_env_rebuild"]["status"] == "skipped"
assert "ensure_plugin_code_env" in res["code_env_rebuild"]["hint"]
assert "whether the plugin needs" in res["code_env_rebuild"]["hint"]
def test_update_plugin_keeps_a_landed_update_when_the_rebuild_fails():
_, client = _installed(
settings={"codeEnvName": "plugin_geocoder"},
futures={
"update": FakeFuture("F2", [_done()]),
"update_code_env": FakeFuture(
"F3", [_code_env_done("plugin_geocoder", error="pip resolution failed")]
),
},
)
with _patch_client(client):
res = _load(
tools.update_plugin(
"store", FakeContext(), plugin_id="geocoder", rebuild_code_env=True
)
)
assert res["status"] == "completed"
assert res["code_env_rebuild"]["status"] == "failed"
assert "pip resolution failed" in res["code_env_rebuild"]["error"]
assert "Dataiku UI" in res["code_env_rebuild"]["hint"]
assert "recreate" in res["code_env_rebuild"]["hint"]
assert "update_plugin" not in res["code_env_rebuild"]["hint"]
@pytest.mark.parametrize("stage", ["inspect", "start", "poll", "terminal"])
def test_update_plugin_keeps_landed_update_after_rebuild_exception(monkeypatch, stage):
from requests.exceptions import Timeout
future = FakeFuture("F3", [_alive()])
plugin, client = _installed(
settings={"codeEnvName": "plugin_geocoder"},
listing_after_change=[_listing_entry("geocoder", version="1.4.0")],
futures={"update": FakeFuture("F2", [_done()]), "update_code_env": future},
)
def fail():
raise Timeout("sensitive transport detail")
if stage == "inspect":
monkeypatch.setattr(plugin, "get_settings", fail)
elif stage == "start":
monkeypatch.setattr(plugin, "update_code_env", fail)
elif stage == "poll":
monkeypatch.setattr(future, "get_state", fail)
else:
future.state = {"alive": False, "aborted": True, "hasResult": False}
with _patch_client(client):
result = _load(
tools.update_plugin(
"store",
FakeContext(),
plugin_id="geocoder",
rebuild_code_env=True,
)
)
assert result["status"] == "completed"
assert result["plugin"]["version"] == "1.4.0"
rebuild = result["code_env_rebuild"]
assert rebuild["status"] == "unknown"
assert "Do not repeat update_plugin" in rebuild["hint"]
assert "sensitive" not in json.dumps(result)
if stage != "inspect":
assert rebuild["code_env_name"] == "plugin_geocoder"
if stage in ("poll", "terminal"):
assert rebuild["future_id"] == "F3"
assert "get_future_status" in rebuild["hint"]
else:
assert "future_id" not in rebuild
def test_update_plugin_timeout_says_the_requested_rebuild_never_started():
"""Absent code_env_rebuild would read as a rebuild that quietly succeeded."""
_, client = _installed(
settings={"codeEnvName": "plugin_geocoder"},
futures={"update": FakeFuture("F2", [_alive(), _alive()])},
)
with _patch_client(client), _exhausted_clock() as mock_time:
mock_time.monotonic.side_effect = incrementing_monotonic()
res = _load(
tools.update_plugin(
"store", FakeContext(), plugin_id="geocoder", rebuild_code_env=True
)
)
assert res["status"] == "still_running"
assert res["code_env_rebuild"]["status"] == "not_started"
assert res["future_id"] == "F2"
assert "get_future_status" in res["code_env_rebuild"]["hint"]
assert "Dataiku UI" in res["code_env_rebuild"]["hint"]
assert "update_plugin" not in res["code_env_rebuild"]["hint"]
def test_update_plugin_from_local_path_targets_the_manifest_id(tmp_path):
directory = _plugin_directory(tmp_path)
plugin, client = _installed(
plugin_id="my-plugin",
version="0.1.0",
futures={"update": FakeFuture(None, [_done()])},
)
with _patch_client(client):
res = _load(
tools.update_plugin("local_path", FakeContext(), local_path=str(directory))
)
assert res["status"] == "completed"
assert _members(plugin.uploaded) == [_MANIFEST, "python-lib/core.py"]
# --------------------------------------------------------------------------- #
# delete_plugin
# --------------------------------------------------------------------------- #
def test_delete_plugin_refuses_while_components_are_in_use():
plugin, client = _installed(
usages={
"usages": [
{
"elementKind": "custom-recipes",
"elementType": "CustomCode_geocode",
"objectType": "RECIPE",
"objectId": "geocode",
"projectKey": "PROJ",
}
],
"missingTypes": [],
}
)
with _patch_client(client):
res = _load(tools.delete_plugin("geocoder", FakeContext()))
assert res["status"] == "refused"
assert res["deleted"] is False
assert res["usage_count"] == 1
assert res["usages"]["columns"][:3] == ["projectKey", "objectType", "objectId"]
assert res["usages"]["rows"][0][:3] == ["PROJ", "RECIPE", "geocode"]
assert plugin.delete_calls == []
def test_delete_plugin_caps_the_reported_usages_but_not_the_count():
usages = [
{"projectKey": f"P{i}", "objectType": "RECIPE", "objectId": f"r{i}"}
for i in range(120)
]
_, client = _installed(usages={"usages": usages, "missingTypes": []})
with _patch_client(client):
res = _load(tools.delete_plugin("geocoder", FakeContext()))
assert res["usage_count"] == 120
assert res["blocker_count"] == 120
assert len(res["usages"]["rows"]) == tools._MAX_REPORTED_USAGES
assert "list_plugin_usages" in res["hint"]
def test_delete_plugin_ignores_instance_wide_unresolvable_component_noise():
"""Dataiku reports missingTypes for the whole instance, not for this plugin."""
_, client = _installed(
usages={
"usages": [],
"missingTypes": [
{"missingType": "Formula", "pluginId": "unknown", "objectId": "r1"},
{"missingType": "geocode", "pluginId": "geocoder", "objectId": "r2"},
],
}
)
with _patch_client(client):
blocked = _load(tools.delete_plugin("geocoder", FakeContext()))
assert blocked["deleted"] is False
assert [item["objectId"] for item in blocked["unresolvable_components"]] == ["r2"]
def test_delete_plugin_refuses_legibly_when_the_usage_analysis_raises():
"""Dataiku raises here for a plugin installed since the last backend reload."""
plugin, client = _installed()
plugin.usages_error = RuntimeError(_UNRESOLVED)
with _patch_client(client):
res = _load(tools.delete_plugin("geocoder", FakeContext()))
assert res["status"] == "refused"
assert res["deleted"] is False
assert res["reason"].startswith("Types were missing")
assert "com.dataiku" not in json.dumps(res)
assert "force=true" in res["hint"]
assert plugin.delete_calls == []
def test_delete_plugin_refuses_legibly_when_the_delete_endpoint_raises():
"""The same check runs server-side on delete, so a clean analysis is not enough."""
plugin, client = _installed(futures={"delete": FakeFuture(None, [_done()])})
plugin.delete_error = RuntimeError(_UNRESOLVED)
with _patch_client(client):
refused = _load(tools.delete_plugin("geocoder", FakeContext()))
forced = _load(tools.delete_plugin("geocoder", FakeContext(), force=True))
assert refused["status"] == "refused"
assert "com.dataiku" not in json.dumps(refused)
assert forced["deleted"] is True
assert plugin.delete_calls == [False, True]
@pytest.mark.parametrize("failing_step", ["usages", "delete"])
def test_delete_plugin_does_not_offer_force_for_an_unrelated_failure(failing_step):
"""A permission error is not a force candidate, from either pre-delete step."""
plugin, client = _installed()
setattr(
plugin,
f"{failing_step}_error",
RuntimeError("Forbidden: you may not manage plugins"),
)
with _patch_client(client), pytest.raises(RuntimeError, match="Forbidden"):
asyncio.run(tools.delete_plugin("geocoder", FakeContext()))
def test_delete_plugin_deletes_when_nothing_uses_it():
plugin, client = _installed(
futures={"delete": FakeFuture("F6", [_alive(), _done(needsRestart=True)])}
)
with _patch_client(client):
res = _load(tools.delete_plugin("geocoder", FakeContext()))
assert res == {
"status": "completed",
"operation": "delete",
"plugin_id": "geocoder",
"deleted": True,
"forced": False,
"needs_reload": False,
"needs_restart": True,
}
assert plugin.delete_calls == [False]
def test_delete_plugin_force_deletes_despite_usages():
plugin, client = _installed(
usages={
"usages": [{"objectId": "geocode", "projectKey": "PROJ"}],
"missingTypes": [],
},
futures={"delete": FakeFuture(None, [_done()])},
)
with _patch_client(client):
res = _load(tools.delete_plugin("geocoder", FakeContext(), force=True))
assert res["deleted"] is True
assert res["forced"] is True
assert plugin.delete_calls == [True]
def test_delete_plugin_does_not_claim_deletion_on_a_failed_result():
_, client = _installed(
futures={
"delete": FakeFuture(
None, [_done(success=False, error={"message": "plugin is locked"})]
)
}
)
with _patch_client(client), pytest.raises(RuntimeError, match="plugin is locked"):
asyncio.run(tools.delete_plugin("geocoder", FakeContext()))
def test_delete_plugin_timeout_does_not_claim_deletion():
_, client = _installed(futures={"delete": FakeFuture("F7", [_alive(), _alive()])})
with _patch_client(client), _exhausted_clock() as mock_time:
mock_time.monotonic.side_effect = incrementing_monotonic()
res = _load(tools.delete_plugin("geocoder", FakeContext()))
assert res["status"] == "still_running"
assert res["deleted"] is False
assert res["future_id"] == "F7"
# --------------------------------------------------------------------------- #
# Shared future and boundary behaviour
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"call",
[
lambda: tools.update_plugin(
"store", FakeContext(), plugin_id="x", timeout_seconds=0
),
lambda: tools.update_plugin(
"store", FakeContext(), plugin_id="x", timeout_seconds=7200
),
lambda: tools.delete_plugin("x", FakeContext(), timeout_seconds=0),
],
)
def test_timeout_bounds_are_validated(call):
with pytest.raises(ValueError, match="timeout_seconds"):
asyncio.run(call())
def test_start_response_without_a_liveness_flag_is_polled_not_trusted():
"""A bare {"jobId": ...} start response must not read as a completed update."""
_, client = _installed(
futures={
"update": FakeFuture("F10", [{"jobId": "F10"}, _done(needsRestart=True)])
}
)
with _patch_client(client):
res = _load(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
assert res["status"] == "completed"
assert res["needs_restart"] is True
def test_future_that_stops_without_a_result_is_not_reported_as_success():
_, client = _installed(
futures={
"update": FakeFuture(
"F11",
[
{"alive": True, "hasResult": False},
{"alive": False, "hasResult": False},
],
)
}
)
with _patch_client(client), pytest.raises(RuntimeError, match="without reporting"):
asyncio.run(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
def test_aborted_future_is_not_reported_as_success():
_, client = _installed(
futures={
"update": FakeFuture(
"F8", [{"alive": False, "aborted": True, "hasResult": False}]
)
}
)
with _patch_client(client), pytest.raises(RuntimeError, match="aborted"):
asyncio.run(tools.update_plugin("store", FakeContext(), plugin_id="geocoder"))
@pytest.mark.parametrize(
("operation", "expected"),
[
("update", "aborted the update of plugin"),
("delete", "aborted the deletion of plugin"),
("rebuild_code_env", "aborted the code environment rebuild of plugin"),
],
)
def test_error_prose_reads_grammatically_for_every_operation(operation, expected):
with pytest.raises(RuntimeError, match=expected):
tools._terminal_result(operation, "geo", {"aborted": True})
def test_poll_interval_scales_with_the_wait_budget(monkeypatch):
"""A ten-minute code-env budget must not be polled every two seconds."""
monkeypatch.setattr(tools, "_MIN_POLL_INTERVAL_SECONDS", 2)
monkeypatch.setattr(tools, "_MAX_POLL_INTERVAL_SECONDS", 30)
slept = []
async def _record(seconds):
slept.append(seconds)
monkeypatch.setattr(tools.asyncio, "sleep", _record)
future = FakeFuture("F1", [_alive(), _alive(), _done()])
asyncio.run(tools._wait_for_future(future, 600))
assert slept and all(interval == 10 for interval in slept)
@pytest.mark.parametrize("http", [False, True])
def test_install_plugin_from_store_reports_completion_and_restart_need(
monkeypatch, http
):
monkeypatch.setattr(request, "is_http_request", lambda: http)
_, client = _absent(
version="1.3.1",
install=FakeFuture(
"F1", [_alive(), _done(needsRestart=True, needsReload=True)]
),
)
with _patch_client(client):
res = _load(tools.install_plugin("store", FakeContext(), plugin_id="geocoder"))
assert res["status"] == "completed"
assert res["operation"] == "install"
assert res["source"] == "store"
assert res["needs_restart"] is True
assert res["needs_reload"] is True
assert res["plugin"]["version"] == "1.3.1"
assert client.store_install_id == "geocoder"
def test_install_plugin_refuses_an_already_installed_plugin():
_, client = _installed(version="1.3.1")
with _patch_client(client), pytest.raises(ValueError) as excinfo:
asyncio.run(tools.install_plugin("store", FakeContext(), plugin_id="geocoder"))
assert "update_plugin" in str(excinfo.value)
# --------------------------------------------------------------------------- #
def test_local_install_is_disabled_in_http_before_any_io(monkeypatch):
monkeypatch.setattr(request, "is_http_request", lambda: True)
monkeypatch.setattr(
tools, "run_blocking", lambda *args: pytest.fail("must fail before IO")
)
with pytest.raises(ValueError, match="unavailable in HTTP"):
_load(
tools.install_plugin(
"local_path", FakeContext(), local_path="/server/plugin.zip"
)
)
@pytest.mark.parametrize(
("directory", "language"), [("python", "python"), ("R", "r"), (None, None)]
)
@pytest.mark.parametrize("wrapped", [False, True])
def test_install_reads_code_env_descriptors_in_normalized_archive(
tmp_path, directory, language, wrapped
):
archive = tmp_path / "sample.zip"
prefix = "wrapper/" if wrapped else ""
with zipfile.ZipFile(archive, "w") as output:
output.writestr(prefix + "plugin.json", json.dumps({"id": "sample"}))
if directory:
output.writestr(prefix + f"code-env/{directory}/desc.json", "{}")
_, client = _absent("sample", {"install": FakeFuture(None, [_done()])})
with _patch_client(client):
result = _load(
tools.install_plugin("local_path", FakeContext(), local_path=str(archive))
)
assert result["code_env"]["status"] == ("required" if language else "not_required")
assert result["code_env"]["languages"] == ([language] if language else [])
def _creation(client=None, states=None):
if client is None:
plugin, client = _installed()
else:
plugin = client.get_plugin("geocoder")
future = FakeFuture("F4", states or [_code_env_done("plugin_geocoder_managed")])
plugin.futures["create_code_env"] = future
return plugin, client, future
def test_ensure_plugin_code_env_creates_and_verifies_binding():
plugin, client, future = _creation()
with _patch_client(client):
result = _load(
tools.ensure_plugin_code_env(
"geocoder", FakeContext(), python_interpreter="PYTHON311"
)
)
assert result["status"] == "completed"
assert result["build"]["status"] == "completed"
assert plugin.settings.get_raw()["codeEnvName"] == result["code_env_name"]
assert plugin.created_interpreter == "PYTHON311"
def test_plugin_interpreter_schema_explains_dataiku_selection():
registered = asyncio.run(tools.mcp.list_tools())
tool = next(item for item in registered if item.name == "ensure_plugin_code_env")
interpreter = tool.parameters["properties"]["python_interpreter"]
assert "accepted versions" in interpreter["description"]
assert "server availability" in interpreter["description"]
assert interpreter["default"] is None
@pytest.mark.parametrize(
"message",
[
"com.dataiku.dip.exceptions.CodedException: ERR_PLUGIN_WITHOUT_CODEENV: no specification",
"com.dataiku.dip.exceptions.CodedException: The plugin geocoder doesn't specify a code env, cannot create it",
],
)
def test_store_without_code_env_is_not_required(message):
from dataikuapi.utils import DataikuException
plugin, client, _ = _creation()
plugin.create_error = DataikuException(message)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["status"] == "not_required"
assert "codeEnvName" not in plugin.settings.get_raw()
@pytest.mark.parametrize(
"message",
[
"permission denied",
"ERR_PLUGIN_WITHOUT_CODEENV_EXTRA",
"cannot create code env for another reason",
],
)
def test_other_creation_errors_remain_errors(message):
from dataikuapi.utils import DataikuException
plugin, client, _ = _creation()
plugin.create_error = DataikuException(message)
with _patch_client(client), pytest.raises(RuntimeError, match="refused") as excinfo:
_load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert message in str(excinfo.value)
assert "codeEnvName" not in plugin.settings.get_raw()
def test_failed_build_is_preserved_beside_binding():
plugin, client, _ = _creation(
states=[_code_env_done("plugin_geocoder_managed", error="pip install failed")]
)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["build"]["status"] == "failed"
assert result["code_env_name"] == plugin.settings.get_raw()["codeEnvName"]
def test_bound_environment_does_not_create_or_claim_build_success():
plugin, client = _installed(settings={"codeEnvName": "already-bound"})
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert not result["created"]
assert result["build"]["status"] == "unknown"
assert not plugin.futures
def _with_orphan(**kwargs):
return _installed(
code_envs=[
{"envName": "plugin_geocoder_managed", "deploymentMode": "PLUGIN_MANAGED"}
],
**kwargs,
)
def test_orphan_is_listed_instead_of_guessed_or_duplicated():
plugin, client = _with_orphan()
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["status"] == "recovery_required"
assert result["code_env_names"] == ["plugin_geocoder_managed"]
assert "code_env_name" in result["hint"]
assert not plugin.futures
assert "codeEnvName" not in plugin.settings.get_raw()
def test_named_orphan_is_bound_without_claiming_a_build():
plugin, client = _with_orphan()
with _patch_client(client):
result = _load(
tools.ensure_plugin_code_env(
"geocoder", FakeContext(), code_env_name="plugin_geocoder_managed"
)
)
assert result["status"] == "completed"
assert result["created"] is False
assert result["build"]["status"] == "unknown"
assert plugin.settings.get_raw()["codeEnvName"] == "plugin_geocoder_managed"
assert not plugin.futures
@pytest.mark.parametrize(
"name", ["plugin_other_managed", "plugin_geocoder_managed_1", "my-env"]
)
def test_named_environment_must_be_an_existing_managed_env_of_the_plugin(name):
plugin, client = _with_orphan()
with _patch_client(client), pytest.raises(ValueError, match="Candidates"):
_load(
tools.ensure_plugin_code_env("geocoder", FakeContext(), code_env_name=name)
)
assert "codeEnvName" not in plugin.settings.get_raw()
@pytest.mark.parametrize(
"kwargs",
[{"code_env_name": "plugin_geocoder_managed"}, {"python_interpreter": "PYTHON311"}],
)
def test_bound_plugin_refuses_a_different_environment_request(kwargs):
plugin, client = _installed(settings={"codeEnvName": "already-bound"})
with _patch_client(client), pytest.raises(ValueError, match="already bound"):
_load(tools.ensure_plugin_code_env("geocoder", FakeContext(), **kwargs))
assert plugin.settings.get_raw()["codeEnvName"] == "already-bound"
def test_timeout_is_resumed_by_exact_future_and_then_bound(monkeypatch):
plugin, client, future = _creation(states=[_alive(), _alive()])
with _patch_client(client), _exhausted_clock() as mock_time:
mock_time.monotonic.side_effect = incrementing_monotonic()
result = _load(
tools.ensure_plugin_code_env("geocoder", FakeContext(), timeout_seconds=1)
)
assert result["status"] == "still_running"
assert result["future_id"] == "F4"
assert "codeEnvName" not in plugin.settings.get_raw()
resumed = _build_future(
"F4", "plugin_geocoder_managed", [_code_env_done("plugin_geocoder_managed")]
)
monkeypatch.setattr(tools, "DSSFuture", lambda client, future_id: resumed)
with _patch_client(client):
result = _load(
tools.ensure_plugin_code_env("geocoder", FakeContext(), future_id="F4")
)
assert result["build"]["status"] == "completed"
assert result["created"] is True
assert plugin.settings.get_raw()["codeEnvName"] == "plugin_geocoder_managed"
@pytest.mark.parametrize(
"peeked",
[
{
"payload": {
"action": "import_env",
"displayName": "Import env: plugin_other_managed",
}
},
{"payload": {"action": "update_plugin", "displayName": "Update: geocoder"}},
{"jobId": "wrong"},
],
)
def test_resuming_another_operations_future_neither_reads_nor_binds(
monkeypatch, peeked
):
plugin, client = _installed()
other = FakeFuture("wrong", [_code_env_done("plugin_other_managed")])
other.peeked = peeked
monkeypatch.setattr(tools, "DSSFuture", lambda *args: other)
with (
_patch_client(client),
pytest.raises(ValueError, match="not a code environment build"),
):
_load(
tools.ensure_plugin_code_env("geocoder", FakeContext(), future_id="wrong")
)
assert not getattr(other, "read", False)
assert "codeEnvName" not in plugin.settings.get_raw()
def test_forgotten_future_points_to_named_binding(monkeypatch):
from dataikuapi.utils import DataikuException
_, client = _installed()
gone = FakeFuture("F4", [_alive()])
def forgotten():
raise DataikuException(
"java.lang.IllegalArgumentException: No future by that id"
)
gone.peek_state = forgotten
monkeypatch.setattr(tools, "DSSFuture", lambda *args: gone)
with _patch_client(client), pytest.raises(ValueError, match="code_env_name"):
_load(tools.ensure_plugin_code_env("geocoder", FakeContext(), future_id="F4"))
def test_binding_failure_reports_the_reason_and_the_name_to_bind(monkeypatch):
from dataikuapi.utils import DataikuException
plugin, client, _ = _creation()
def fail_save():
raise DataikuException(
"com.dataiku.dip.exceptions.UnauthorizedException: Action forbidden"
)
monkeypatch.setattr(plugin.settings, "save", fail_save)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["status"] == "binding_failed"
assert result["code_env_name"] == "plugin_geocoder_managed"
assert result["error"] == "Action forbidden"
def test_changed_binding_is_preserved(monkeypatch):
plugin, client, future = _creation(
states=[_alive(), _code_env_done("plugin_geocoder_managed")]
)
original = future.get_state
def switched():
plugin.settings.raw["codeEnvName"] = "chosen-elsewhere"
return original()
monkeypatch.setattr(future, "get_state", switched)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["status"] == "binding_changed"
assert result["bound_code_env_name"] == "chosen-elsewhere"
assert plugin.settings.raw["codeEnvName"] == "chosen-elsewhere"
@pytest.mark.parametrize("no_spec", [True, False])
def test_code_env_contract_through_real_sdk_http_serialization(monkeypatch, no_spec):
"""The real SDK drops errorCode; exercise that boundary instead of a fake exception."""
import requests
from dataikuapi import DSSClient
client = DSSClient("https://synthetic.example.com", "synthetic-key")
settings = {}
calls = []
def respond(method, url, **kwargs):
path = url.split("/dip/publicapi", 1)[1]
calls.append((method, path))
status = 200
if path == "/plugins/":
value = [_listing_entry("geocoder")]
elif path == "/admin/code-envs/":
value = []
elif path.endswith("/settings"):
if method == "POST":
settings.update(json.loads(kwargs["data"]))
value = dict(settings)
elif path.endswith("/code-env/actions/create"):
body = json.loads(kwargs["data"])
assert body["deploymentMode"] == "PLUGIN_MANAGED"
assert body["pythonInterpreter"] is None
if no_spec:
status = 400
value = {
"errorType": "com.dataiku.dip.exceptions.CodedException",
"errorCode": "ERR_PLUGIN_WITHOUT_CODEENV",
"detailedMessage": "The plugin geocoder doesn't specify a code env, cannot create it",
}
else:
value = _code_env_done("plugin_geocoder_managed")
else:
pytest.fail(f"Unexpected SDK call {method} {path}")
response = requests.Response()
response.status_code = status
response._content = json.dumps(value).encode()
return response
monkeypatch.setattr(client._session, "request", respond)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["status"] == ("not_required" if no_spec else "completed")
assert ("codeEnvName" in settings) is not no_spec
assert ("POST", "/plugins/geocoder/settings") in calls if not no_spec else True
def test_environment_name_without_build_report_is_not_verified_success():
_, client, _ = _creation(
states=[{"hasResult": True, "result": {"envName": "plugin_geocoder_managed"}}]
)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["build"]["status"] == "unknown"
@pytest.mark.parametrize(
"name",
[
"plugin_geocoder-extra_managed",
"plugin_geocoder_managed_extra",
"plugin_other_managed",
],
)
def test_similarly_named_environments_do_not_block_initial_creation(name):
_, client = _installed(
code_envs=[{"envName": name, "deploymentMode": "PLUGIN_MANAGED"}]
)
_, client, _ = _creation(client)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["created"] is True
_GIT_URL = "https://git.example.com/team/plugin.git"
def _git_done(**kwargs):
return _done(pluginDesc={"desc": {"id": "geocoder"}}, **kwargs)
@pytest.mark.parametrize("wait", [False, True])
def test_git_install_keeps_future_when_plugin_id_is_not_known(wait):
_, client = _absent(install=FakeFuture("G2", [_alive(), _alive()]))
with _patch_client(client), _exhausted_clock() as mock_time:
mock_time.monotonic.side_effect = incrementing_monotonic()
result = _load(
tools.install_plugin(
"git",
FakeContext(),
repository_url=_GIT_URL,
checkout="main",
wait_for_completion=wait,
timeout_seconds=1,
)
)
assert result["status"] == ("still_running" if wait else "started")
assert result["future_id"] == "G2"
assert "plugin_id" not in result
def test_git_install_reports_uncertain_verification_without_restarting():
_, client = _absent(install=FakeFuture(None, [_git_done()]))
client.listing_after_change = []
with _patch_client(client):
result = _load(
tools.install_plugin(
"git",
FakeContext(),
repository_url=_GIT_URL,
checkout="main",
)
)
assert result["status"] == "verification_required"
assert "list_plugins" in result["hint"]
@pytest.mark.parametrize("operation", ["install", "update"])
def test_git_contract_with_real_sdk(monkeypatch, operation):
import requests
from dataikuapi import DSSClient
monkeypatch.setattr(request, "is_http_request", lambda: True)
client = DSSClient("https://synthetic.example.com", "synthetic-key")
calls = []
endpoint = (
"/plugins/actions/installFromGit"
if operation == "install"
else "/plugins/geocoder/actions/updateFromGit"
)
checkout = "v1.0" if operation == "install" else _GIT_COMMIT
def respond(method, url, **kwargs):
path = url.split("/dip/publicapi", 1)[1]
calls.append((method, path))
if path == endpoint:
assert json.loads(kwargs["data"]) == {
"gitRepositoryUrl": _GIT_URL,
"gitCheckout": checkout,
"gitSubpath": "plugin",
}
value = {"jobId": "G4", **_alive()}
elif path == "/futures/G4":
value = _git_done(needsReload=True)
value["result"]["settings"] = {"secret": "do-not-return"}
elif path == "/plugins/":
value = [_listing_entry("geocoder")]
else:
pytest.fail(f"Unexpected SDK call {method} {path}")
response = requests.Response()
response.status_code = 200
response._content = json.dumps(value).encode()
return response
monkeypatch.setattr(client._session, "request", respond)
with _patch_client(client):
result = _load(
getattr(tools, f"{operation}_plugin")(
"git",
FakeContext(),
repository_url=_GIT_URL,
checkout=checkout,
**({"plugin_id": "geocoder"} if operation == "update" else {}),
subpath="plugin",
)
)
assert result["status"] == "completed"
assert result["plugin_id"] == result["plugin"]["id"] == "geocoder"
assert result["needs_reload"] is True
if operation == "install":
assert result["code_env"] == {"status": "unknown"}
assert "do-not-return" not in json.dumps(result)
assert calls == ([("GET", "/plugins/")] if operation == "update" else []) + [
("POST", endpoint),
("GET", "/futures/G4"),
("GET", "/plugins/"),
]
_GIT_COMMIT = "a1" * 20
@pytest.mark.parametrize(
"http,rebuild,checkout,subpath",
[
(True, True, "release/1.0", "plugins/geocoder"),
(False, False, "v2.0", None),
(False, False, _GIT_COMMIT, None),
],
)
def test_git_update_uses_chosen_revision_and_path(
monkeypatch, http, rebuild, checkout, subpath
):
monkeypatch.setattr(request, "is_http_request", lambda: http)
plugin, client = _installed(
settings={"codeEnvName": "plugin_geocoder_managed"},
futures={
"update": FakeFuture("GU1", [_alive(), _git_done(needsReload=True)]),
"update_code_env": FakeFuture(
None, [_code_env_done("plugin_geocoder_managed")]
),
},
listing_after_change=[_listing_entry("geocoder", "2.0.0")],
)
with _patch_client(client):
result = _load(
tools.update_plugin(
"git",
FakeContext(),
plugin_id="geocoder",
repository_url=_GIT_URL,
checkout=checkout,
subpath=subpath,
rebuild_code_env=rebuild,
)
)
assert plugin.git_update_args == (_GIT_URL, checkout, subpath)
assert result["plugin"]["version"] == "2.0.0"
assert result["needs_reload"] is True
assert result["status"] == "completed"
assert ("code_env_rebuild" in result) is rebuild
if rebuild:
assert result["code_env_rebuild"]["status"] == "completed"
@pytest.mark.parametrize("wait", [False, True])
def test_git_update_preserves_future_and_defers_rebuild(wait):
_, client = _installed(futures={"update": FakeFuture("GU2", [_alive(), _alive()])})
with _patch_client(client), _exhausted_clock() as mock_time:
mock_time.monotonic.side_effect = incrementing_monotonic()
result = _load(
tools.update_plugin(
"git",
FakeContext(),
plugin_id="geocoder",
repository_url=_GIT_URL,
checkout=_GIT_COMMIT,
wait_for_completion=wait,
rebuild_code_env=wait,
timeout_seconds=1,
)
)
assert result["future_id"] == "GU2"
assert result["status"] == ("still_running" if wait else "started")
if wait:
assert result["code_env_rebuild"]["status"] == "not_started"
@pytest.mark.parametrize("actual_id", [None, "other-plugin"])
def test_git_update_unexpected_result_requires_verification_and_skips_rebuild(
actual_id,
):
_, client = _installed(
futures={
"update": FakeFuture(None, [_done(pluginDesc={"desc": {"id": actual_id}})])
}
)
with _patch_client(client):
result = _load(
tools.update_plugin(
"git",
FakeContext(),
plugin_id="geocoder",
repository_url=_GIT_URL,
checkout=_GIT_COMMIT,
rebuild_code_env=True,
)
)
assert result["status"] == "verification_required"
assert "code_env_rebuild" not in result
assert result.get("reported_plugin_id") == actual_id
def test_git_update_refuses_missing_target_before_mutation():
_, client = _absent()
with _patch_client(client), pytest.raises(ValueError, match="not installed"):
_load(
tools.update_plugin(
"git",
FakeContext(),
plugin_id="geocoder",
repository_url=_GIT_URL,
checkout=_GIT_COMMIT,
)
)
@pytest.mark.parametrize(
"operation,stage", [("install", "start"), ("install", "poll"), ("update", "result")]
)
def test_git_access_failure_provides_help_and_recovery(monkeypatch, operation, stage):
from dataikuapi.utils import DataikuException
plugin, client = _installed() if operation == "update" else _absent()
future = FakeFuture("G-auth", [_alive()])
futures = plugin.futures if operation == "update" else client.futures
futures[operation] = future
secret = "synthetic-token"
message = f"Authentication failed https://user:{secret}@git.example.com/plugin"
def fail(*args, **kwargs):
raise DataikuException(message)
if stage == "start":
monkeypatch.setattr(client, "install_plugin_from_git", fail)
elif stage == "poll":
monkeypatch.setattr(future, "get_state", fail)
else:
futures[operation] = FakeFuture(
"G-auth", [_done(success=False, installationError={"message": message})]
)
with _patch_client(client), pytest.raises(RuntimeError) as error:
_load(
getattr(tools, f"{operation}_plugin")(
"git",
FakeContext(),
repository_url=_GIT_URL,
checkout="main",
**({"plugin_id": "geocoder"} if operation == "update" else {}),
)
)
assert "Dataiku's Git settings" in str(error.value)
assert "Credentials on the Headless machine do not apply" in str(error.value)
assert secret not in str(error.value)
if stage != "start":
assert "G-auth" in str(error.value)
def test_git_non_auth_failure_does_not_claim_missing_credentials():
_, client = _absent(
install=FakeFuture(
None,
[
_done(
success=False,
installationError={"message": "Remote branch does not exist"},
)
],
)
)
with _patch_client(client), pytest.raises(RuntimeError) as error:
_load(
tools.install_plugin(
"git", FakeContext(), repository_url=_GIT_URL, checkout="missing"
)
)
assert "could not authenticate" not in str(error.value)
assert "checkout" in str(error.value)
assert "list_plugins" in str(error.value)
@pytest.mark.parametrize(
"operation,source,overrides",
[
("install", "git", {"repository_url": None}),
("update", "git", {"checkout": " "}),
("update", "git", {"plugin_id": None}),
("install", "git", {"plugin_id": "geocoder"}),
("update", "git", {"local_path": "/tmp/plugin"}),
("install", "git", {"subpath": "../plugin"}),
("update", "git", {"subpath": "/plugin"}),
(
"install",
"git",
{"repository_url": "https://user:synthetic-token@git.example.com/plugin"},
),
(
"update",
"git",
{"repository_url": "https://git.example.com/plugin?token=synthetic-token"},
),
("install", "git", {"repository_url": "file:///server/plugin"}),
("install", "store", {}),
("update", "local_path", {}),
],
)
def test_git_source_validation_rejects_invalid_arguments_before_io(
operation, source, overrides
):
arguments = dict(repository_url=_GIT_URL, checkout="main")
if operation == "update":
arguments["plugin_id"] = "geocoder"
arguments.update(overrides)
with pytest.raises(ValueError) as error:
_load(getattr(tools, f"{operation}_plugin")(source, FakeContext(), **arguments))
assert "synthetic-token" not in str(error.value)
@pytest.mark.parametrize(
"url",
[
"git@git.example.com:team/plugin.git",
"ssh://git@git.example.com/team/plugin.git",
],
)
def test_git_ssh_remote_forms_are_accepted(url):
assert tools._require_git_repository_url(url) == url
@pytest.mark.parametrize("after_save", [False, True])
def test_binding_transport_timeout_keeps_build_and_supports_retry(
monkeypatch, after_save
):
from requests.exceptions import Timeout
plugin, client, _ = _creation()
original = plugin.settings.save
def save():
if after_save:
original()
else:
plugin.settings.raw.pop("codeEnvName", None)
raise Timeout("sensitive transport detail")
monkeypatch.setattr(plugin.settings, "save", save)
with _patch_client(client):
result = _load(tools.ensure_plugin_code_env("geocoder", FakeContext()))
assert result["status"] == "binding_unknown"
assert result["build"]["status"] == "completed"
assert result["code_env_name"] == "plugin_geocoder_managed"
assert "sensitive" not in json.dumps(result)
assert bool(plugin.settings.raw.get("codeEnvName")) is after_save
client.code_envs = [
{"deploymentMode": "PLUGIN_MANAGED", "envName": result["code_env_name"]}
]
monkeypatch.setattr(plugin.settings, "save", original)
with _patch_client(client):
result = _load(
tools.ensure_plugin_code_env(
"geocoder", FakeContext(), code_env_name=result["code_env_name"]
)
)
assert result["status"] == "completed"
def test_automation_build_resumes_only_for_retained_owner(monkeypatch):
plugin, client, _ = _creation(states=[_alive(), _alive()])
with _patch_client(client), _exhausted_clock() as clock:
clock.monotonic.side_effect = incrementing_monotonic()
_load(
tools.ensure_plugin_code_env("geocoder", FakeContext(), timeout_seconds=1)
)
resumed = _build_future(
"F4", "plugin_geocoder_managed", [_code_env_done("plugin_geocoder_managed")]
)
resumed.peeked["payload"]["displayName"] = "Import env"
monkeypatch.setattr(tools, "DSSFuture", lambda *args: resumed)
with _patch_client(client):
result = _load(
tools.ensure_plugin_code_env("geocoder", FakeContext(), future_id="F4")
)
assert result["status"] == "completed"
assert not tools._BUILD_OWNERS
@pytest.mark.parametrize("scope", ["plugin", "instance", "caller", "restart"])
def test_automation_build_does_not_consume_an_unowned_future(monkeypatch, scope):
_, client = _installed()
key = tools._build_owner_key(client, "F4")
tools._BUILD_OWNERS[key] = "other" if scope == "plugin" else "geocoder"
if scope == "instance":
client.host = "https://other.example.com"
elif scope == "caller":
monkeypatch.setattr(request, "get_request_owner", lambda: ("oidc", "other"))
elif scope == "restart":
tools._BUILD_OWNERS.clear()
future = _build_future(
"F4", "plugin_geocoder_managed", [_code_env_done("plugin_geocoder_managed")]
)
future.peeked["payload"]["displayName"] = "Import env"
monkeypatch.setattr(tools, "DSSFuture", lambda *args: future)
with (
_patch_client(client),
pytest.raises(ValueError, match="not a code environment build"),
):
_load(tools.ensure_plugin_code_env("geocoder", FakeContext(), future_id="F4"))
assert not getattr(future, "read", False)
SHA-256: 002019bb10e95a766452613ef8c972ba959d947cd80eeeed38528b37bfdd982f