mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-11 18:16:40 +00:00
Add test suite for the Bonsai Viewer Autodesk connector
Introduce pytest coverage for the previously untested connector — rpc, cache, settings, autodesk (auth + APS client) and connector handlers — 94 tests, runnable via the new `test` optional-dependency extra. To make HTTP, time and the OAuth redirect testable without a network or real sockets, add dependency-injection seams to autodesk.py: AuthSessionService and ApsClient accept an optional httpx transport; AuthSessionService accepts an injectable clock and callback_waiter; and _wait_for_callback is extracted to the module-level wait_for_oauth_callback. All seams default to the previous behaviour. Remove the APS_CLIENT_ID environment-variable override: the client id now comes solely from settings.json, collapsing settings.load_client_id and simplifying the settings dialog. CI: the build-bonsaiviewer-autodesk workflow gains a `test` job (Python 3.11 + 3.13) that gates the build matrix. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Dion Moult
@@ -12,8 +12,41 @@ on:
|
||||
- '.github/workflows/build-bonsaiviewer-autodesk.yml'
|
||||
|
||||
jobs:
|
||||
test:
|
||||
name: test-py${{ matrix.python-version }}
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
# Exercise the floor and a current version of the supported range
|
||||
# (pyproject requires-python = ">=3.11").
|
||||
python-version: ['3.11', '3.13']
|
||||
|
||||
defaults:
|
||||
run:
|
||||
working-directory: src/bonsaiviewer-autodesk
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
|
||||
- uses: actions/setup-python@v6
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
|
||||
# The connector imports tkinter/customtkinter (via the test suite's
|
||||
# connector coverage), so fail loudly here if Tk is missing.
|
||||
- name: Verify tkinter is available
|
||||
run: python -c "import tkinter; print('Tk', tkinter.TkVersion)"
|
||||
|
||||
- name: Install package and test deps
|
||||
run: python -m pip install ".[test]"
|
||||
|
||||
- name: Run pytest
|
||||
run: python -m pytest -q
|
||||
|
||||
build:
|
||||
name: ${{ matrix.os_label }}-${{ matrix.arch }}
|
||||
needs: test
|
||||
runs-on: ${{ matrix.runner }}
|
||||
strategy:
|
||||
fail-fast: false
|
||||
|
||||
@@ -22,6 +22,13 @@ from bonsaiviewer_autodesk.rpc import JSONRPC_INTERNAL_ERROR, RpcError
|
||||
|
||||
Progress = Callable[[str, str, "int | None"], None]
|
||||
|
||||
# (host, port, path, expected_state) -> authorization code
|
||||
CallbackWaiter = Callable[[str, int, str, str], str]
|
||||
|
||||
|
||||
def _utcnow() -> dt.datetime:
|
||||
return dt.datetime.now(dt.timezone.utc)
|
||||
|
||||
|
||||
def _no_keyring_error() -> RpcError:
|
||||
return RpcError(
|
||||
@@ -93,6 +100,48 @@ def _noop_progress(_phase: str, _message: str, _percent: int | None = None) -> N
|
||||
return
|
||||
|
||||
|
||||
def wait_for_oauth_callback(host: str, port: int, path: str, expected_state: str) -> str:
|
||||
"""Block on a single OAuth redirect to ``http://host:port/path`` and return
|
||||
the authorization code. Raises ``RpcError`` on an OAuth error, a ``state``
|
||||
mismatch, or a missing code. This is the default ``callback_waiter`` for
|
||||
:class:`AuthSessionService`; tests inject a stub instead."""
|
||||
result: dict[str, str] = {}
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def do_GET(self) -> None:
|
||||
parsed = urllib.parse.urlparse(self.path)
|
||||
if parsed.path != path:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
query = urllib.parse.parse_qs(parsed.query)
|
||||
result["state"] = query.get("state", [""])[0]
|
||||
result["code"] = query.get("code", [""])[0]
|
||||
result["error"] = query.get("error", [""])[0]
|
||||
body = b"<html><body><h2>Authentication complete. You can close this window.</h2></body></html>"
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, format: str, *args: object) -> None:
|
||||
return
|
||||
|
||||
server = HTTPServer((host, port), Handler)
|
||||
server.handle_request()
|
||||
server.server_close()
|
||||
|
||||
if result.get("error"):
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, f"Autodesk returned OAuth error '{result['error']}'.")
|
||||
if result.get("state") != expected_state:
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, "OAuth state mismatch.")
|
||||
code = result.get("code", "")
|
||||
if not code:
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, "OAuth callback did not return an authorization code.")
|
||||
return code
|
||||
|
||||
|
||||
class AuthSessionService:
|
||||
authorize_endpoint = "https://developer.api.autodesk.com/authentication/v2/authorize"
|
||||
token_endpoint = "https://developer.api.autodesk.com/authentication/v2/token"
|
||||
@@ -104,12 +153,17 @@ class AuthSessionService:
|
||||
callback_url: str,
|
||||
scope: str,
|
||||
token_store: KeyringTokenStore,
|
||||
transport: httpx.BaseTransport | None = None,
|
||||
now: Callable[[], dt.datetime] | None = None,
|
||||
callback_waiter: CallbackWaiter | None = None,
|
||||
) -> None:
|
||||
self.client_id = client_id
|
||||
self.callback_url = callback_url
|
||||
self.scope = scope
|
||||
self.token_store = token_store
|
||||
self.http = httpx.Client(timeout=60)
|
||||
self.http = httpx.Client(timeout=60, transport=transport)
|
||||
self._now = now or _utcnow
|
||||
self._callback_waiter = callback_waiter or wait_for_oauth_callback
|
||||
|
||||
def get_token(self) -> StoredToken | None:
|
||||
raw = self.token_store.load()
|
||||
@@ -117,7 +171,7 @@ class AuthSessionService:
|
||||
|
||||
def ensure_access_token(self, progress: Progress = _noop_progress) -> str:
|
||||
token = self.get_token()
|
||||
now = dt.datetime.now(dt.timezone.utc)
|
||||
now = self._now()
|
||||
if token and token.access_token_expires_at > now + dt.timedelta(minutes=1):
|
||||
return token.access_token
|
||||
if token and token.refresh_token_expires_at > now + dt.timedelta(minutes=1):
|
||||
@@ -148,7 +202,7 @@ class AuthSessionService:
|
||||
|
||||
progress("auth", "Opening browser for Autodesk sign-in", None)
|
||||
webbrowser.open(authorize_url)
|
||||
code = self._wait_for_callback(
|
||||
code = self._callback_waiter(
|
||||
callback.hostname or "127.0.0.1",
|
||||
callback.port or 80,
|
||||
callback.path or "/",
|
||||
@@ -192,7 +246,7 @@ class AuthSessionService:
|
||||
return refreshed
|
||||
|
||||
def _token_from_payload(self, payload: dict[str, Any]) -> StoredToken:
|
||||
now = dt.datetime.now(dt.timezone.utc)
|
||||
now = self._now()
|
||||
refresh_ttl = int(payload.get("refresh_token_expires_in", 15 * 24 * 60 * 60))
|
||||
return StoredToken(
|
||||
client_id=self.client_id,
|
||||
@@ -203,48 +257,10 @@ class AuthSessionService:
|
||||
scope=self.scope,
|
||||
)
|
||||
|
||||
def _wait_for_callback(self, host: str, port: int, path: str, expected_state: str) -> str:
|
||||
result: dict[str, str] = {}
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def do_GET(self) -> None:
|
||||
parsed = urllib.parse.urlparse(self.path)
|
||||
if parsed.path != path:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
query = urllib.parse.parse_qs(parsed.query)
|
||||
result["state"] = query.get("state", [""])[0]
|
||||
result["code"] = query.get("code", [""])[0]
|
||||
result["error"] = query.get("error", [""])[0]
|
||||
body = b"<html><body><h2>Authentication complete. You can close this window.</h2></body></html>"
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, format: str, *args: object) -> None:
|
||||
return
|
||||
|
||||
server = HTTPServer((host, port), Handler)
|
||||
server.handle_request()
|
||||
server.server_close()
|
||||
|
||||
if result.get("error"):
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, f"Autodesk returned OAuth error '{result['error']}'.")
|
||||
if result.get("state") != expected_state:
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, "OAuth state mismatch.")
|
||||
code = result.get("code", "")
|
||||
if not code:
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, "OAuth callback did not return an authorization code.")
|
||||
return code
|
||||
|
||||
|
||||
class ApsClient:
|
||||
def __init__(self, auth: AuthSessionService) -> None:
|
||||
def __init__(self, auth: AuthSessionService, *, transport: httpx.BaseTransport | None = None) -> None:
|
||||
self.auth = auth
|
||||
self.http = httpx.Client(timeout=120)
|
||||
self.http = httpx.Client(timeout=120, transport=transport)
|
||||
|
||||
# Browsing -----------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -43,19 +43,9 @@ def _write(data: dict[str, Any]) -> None:
|
||||
_settings_path().write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def stored_client_id() -> str:
|
||||
"""Whatever is persisted in settings.json — ignores the env var."""
|
||||
return str(_read().get("client_id", "")).strip()
|
||||
|
||||
|
||||
def env_client_id() -> str:
|
||||
"""Whatever APS_CLIENT_ID currently has — ignores settings.json."""
|
||||
return os.environ.get("APS_CLIENT_ID", "").strip()
|
||||
|
||||
|
||||
def load_client_id() -> str:
|
||||
"""The effective value: env var wins, so dev overrides keep working."""
|
||||
return env_client_id() or stored_client_id()
|
||||
"""The client id persisted in settings.json, or "" if none is set."""
|
||||
return str(_read().get("client_id", "")).strip()
|
||||
|
||||
|
||||
def save_client_id(client_id: str) -> None:
|
||||
|
||||
@@ -613,9 +613,7 @@ class SettingsDialog(_BaseDialog):
|
||||
super().__init__("Autodesk Connector Settings", size=(520, 400), resizable=False)
|
||||
self.connector = connector
|
||||
|
||||
env_override = settings.env_client_id()
|
||||
stored = settings.stored_client_id()
|
||||
effective = env_override or stored
|
||||
client_id = settings.load_client_id()
|
||||
callback_port = settings.stored_callback_port()
|
||||
|
||||
body = ctk.CTkFrame(self, fg_color="transparent")
|
||||
@@ -638,17 +636,7 @@ class SettingsDialog(_BaseDialog):
|
||||
ctk.CTkLabel(body, text="APS client id", anchor="w").pack(fill="x")
|
||||
self.client_id_entry = ctk.CTkEntry(body, placeholder_text="Paste your APS client id")
|
||||
self.client_id_entry.pack(fill="x", pady=(6, 8))
|
||||
self.client_id_entry.insert(0, stored)
|
||||
|
||||
if env_override:
|
||||
ctk.CTkLabel(
|
||||
body,
|
||||
text=f"APS_CLIENT_ID environment variable is set ({env_override}) and overrides the saved value.",
|
||||
anchor="w",
|
||||
wraplength=460,
|
||||
justify="left",
|
||||
text_color=("#b45309", "#f59e0b"),
|
||||
).pack(fill="x", pady=(0, 8))
|
||||
self.client_id_entry.insert(0, client_id)
|
||||
|
||||
ctk.CTkLabel(body, text="OAuth callback port", anchor="w").pack(fill="x")
|
||||
self.callback_port_entry = ctk.CTkEntry(body, placeholder_text=str(settings.DEFAULT_CALLBACK_PORT))
|
||||
@@ -657,7 +645,7 @@ class SettingsDialog(_BaseDialog):
|
||||
|
||||
self.status_label = ctk.CTkLabel(
|
||||
body,
|
||||
text=f"Signed in as {effective}" if effective else "No client id configured.",
|
||||
text=f"Signed in as {client_id}" if client_id else "No client id configured.",
|
||||
anchor="w",
|
||||
wraplength=460,
|
||||
justify="left",
|
||||
@@ -676,7 +664,7 @@ class SettingsDialog(_BaseDialog):
|
||||
border_width=1,
|
||||
)
|
||||
self.signout_button.grid(row=0, column=0, sticky="w")
|
||||
if not effective:
|
||||
if not client_id:
|
||||
self.signout_button.configure(state="disabled")
|
||||
|
||||
ctk.CTkButton(
|
||||
@@ -705,7 +693,7 @@ class SettingsDialog(_BaseDialog):
|
||||
self._on_close()
|
||||
|
||||
def _sign_out(self) -> None:
|
||||
client_id = settings.env_client_id() or settings.stored_client_id()
|
||||
client_id = settings.load_client_id()
|
||||
if not client_id:
|
||||
return
|
||||
if not confirm(
|
||||
|
||||
@@ -16,12 +16,17 @@ dependencies = [
|
||||
|
||||
[project.optional-dependencies]
|
||||
build = ["pyinstaller>=6.0"]
|
||||
test = ["pytest>=8"]
|
||||
|
||||
[tool.setuptools]
|
||||
include-package-data = true
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["bonsaiviewer_autodesk*"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
|
||||
[project.scripts]
|
||||
bonsaiviewer-autodesk = "bonsaiviewer_autodesk.__main__:main"
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Shared fixtures for the bonsaiviewer-autodesk test suite."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import keyring
|
||||
import keyring.backend
|
||||
import keyring.errors
|
||||
import pytest
|
||||
|
||||
from bonsaiviewer_autodesk import cache as cache_module
|
||||
from bonsaiviewer_autodesk import settings as settings_module
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cache_dir(tmp_path, monkeypatch):
|
||||
"""Redirect ``bonsaiviewer_autodesk.cache`` at an isolated tmp directory.
|
||||
|
||||
Patches ``cache_root`` itself rather than ``XDG_CACHE_HOME`` so the
|
||||
redirect holds on every platform — on macOS/Windows ``cache_root`` ignores
|
||||
the XDG variables. Returns the directory ``cache_root()`` now resolves to.
|
||||
"""
|
||||
root = tmp_path / "cache" / "bonsaiviewer-autodesk"
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
monkeypatch.setattr(cache_module, "cache_root", lambda: root)
|
||||
return root
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_dir(tmp_path, monkeypatch):
|
||||
"""Redirect ``bonsaiviewer_autodesk.settings`` at an isolated tmp directory.
|
||||
|
||||
Patches ``config_root`` directly (platform-independent). Returns the
|
||||
directory ``config_root()`` now resolves to.
|
||||
"""
|
||||
root = tmp_path / "config" / "bonsaiviewer-autodesk"
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
monkeypatch.setattr(settings_module, "config_root", lambda: root)
|
||||
return root
|
||||
|
||||
|
||||
class InMemoryKeyring(keyring.backend.KeyringBackend):
|
||||
"""A keyring backend that keeps secrets in a dict — never touches the OS."""
|
||||
|
||||
priority = 1 # type: ignore[assignment]
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._store: dict[tuple[str, str], str] = {}
|
||||
|
||||
def get_password(self, service: str, username: str) -> str | None:
|
||||
return self._store.get((service, username))
|
||||
|
||||
def set_password(self, service: str, username: str, password: str) -> None:
|
||||
self._store[(service, username)] = password
|
||||
|
||||
def delete_password(self, service: str, username: str) -> None:
|
||||
try:
|
||||
del self._store[(service, username)]
|
||||
except KeyError as exc:
|
||||
raise keyring.errors.PasswordDeleteError("not found") from exc
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def memory_keyring():
|
||||
"""Swap in the in-memory keyring backend for the duration of a test."""
|
||||
backend = InMemoryKeyring()
|
||||
previous = keyring.get_keyring()
|
||||
keyring.set_keyring(backend)
|
||||
try:
|
||||
yield backend
|
||||
finally:
|
||||
keyring.set_keyring(previous)
|
||||
@@ -0,0 +1,676 @@
|
||||
"""Tests for the Autodesk auth + APS client (``bonsaiviewer_autodesk.autodesk``).
|
||||
|
||||
HTTP is mocked through the injectable ``transport`` seam using
|
||||
``httpx.MockTransport``; time through the injectable ``now`` clock; and the
|
||||
OAuth redirect through the injectable ``callback_waiter``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
from collections import deque
|
||||
|
||||
import httpx
|
||||
import keyring.errors
|
||||
import pytest
|
||||
|
||||
from bonsaiviewer_autodesk import autodesk
|
||||
from bonsaiviewer_autodesk.rpc import RpcError
|
||||
|
||||
|
||||
# --- HTTP routing ------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_response(spec: dict) -> httpx.Response:
|
||||
kwargs = {key: spec[key] for key in ("json", "text", "content", "headers") if key in spec}
|
||||
return httpx.Response(spec.get("status", 200), **kwargs)
|
||||
|
||||
|
||||
class Router:
|
||||
"""A tiny httpx.MockTransport router.
|
||||
|
||||
Routes match on HTTP method plus a substring of the request URL, in
|
||||
declaration order. Pass multiple specs to a route to return them in turn
|
||||
(the last one repeats); every request is recorded on ``requests``.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._routes: list[tuple[str, str, deque]] = []
|
||||
self.requests: list[httpx.Request] = []
|
||||
|
||||
def add(self, method: str, contains: str, *specs: dict) -> "Router":
|
||||
self._routes.append((method, contains, deque(specs or ({},))))
|
||||
return self
|
||||
|
||||
def _handle(self, request: httpx.Request) -> httpx.Response:
|
||||
self.requests.append(request)
|
||||
for method, contains, specs in self._routes:
|
||||
if request.method == method and contains in str(request.url):
|
||||
spec = specs[0] if len(specs) == 1 else specs.popleft()
|
||||
return _build_response(spec)
|
||||
return httpx.Response(404, text=f"unrouted {request.method} {request.url}")
|
||||
|
||||
@property
|
||||
def transport(self) -> httpx.MockTransport:
|
||||
return httpx.MockTransport(self._handle)
|
||||
|
||||
def count(self, method: str, contains: str) -> int:
|
||||
return sum(
|
||||
1 for r in self.requests if r.method == method and contains in str(r.url)
|
||||
)
|
||||
|
||||
|
||||
# --- fakes -------------------------------------------------------------------
|
||||
|
||||
|
||||
class FakeTokenStore:
|
||||
"""In-memory stand-in for KeyringTokenStore."""
|
||||
|
||||
def __init__(self, initial: dict | None = None) -> None:
|
||||
self.value = initial
|
||||
|
||||
def load(self) -> dict | None:
|
||||
return self.value
|
||||
|
||||
def save(self, value: dict) -> None:
|
||||
self.value = value
|
||||
|
||||
def delete(self) -> None:
|
||||
self.value = None
|
||||
|
||||
|
||||
class FakeAuth:
|
||||
"""Minimal AuthSessionService stand-in for ApsClient tests."""
|
||||
|
||||
def ensure_access_token(self) -> str:
|
||||
return "fake-token"
|
||||
|
||||
|
||||
FIXED_NOW = dt.datetime(2026, 1, 1, 12, 0, 0, tzinfo=dt.timezone.utc)
|
||||
|
||||
|
||||
def iso(offset_seconds: int) -> str:
|
||||
return (FIXED_NOW + dt.timedelta(seconds=offset_seconds)).isoformat()
|
||||
|
||||
|
||||
def stored_token_dict(*, access_offset: int, refresh_offset: int) -> dict:
|
||||
return {
|
||||
"client_id": "cid",
|
||||
"access_token": "current-access",
|
||||
"refresh_token": "current-refresh",
|
||||
"access_token_expires_at_utc": iso(access_offset),
|
||||
"refresh_token_expires_at_utc": iso(refresh_offset),
|
||||
"scope": "data:read",
|
||||
}
|
||||
|
||||
|
||||
def make_auth(
|
||||
*,
|
||||
router: Router | None = None,
|
||||
token_store: FakeTokenStore | None = None,
|
||||
callback_waiter=None,
|
||||
callback_url: str = "http://localhost:8080/",
|
||||
) -> autodesk.AuthSessionService:
|
||||
return autodesk.AuthSessionService(
|
||||
client_id="cid",
|
||||
callback_url=callback_url,
|
||||
scope="data:read",
|
||||
token_store=token_store or FakeTokenStore(),
|
||||
transport=(router or Router()).transport,
|
||||
now=lambda: FIXED_NOW,
|
||||
callback_waiter=callback_waiter,
|
||||
)
|
||||
|
||||
|
||||
def make_client(router: Router) -> autodesk.ApsClient:
|
||||
return autodesk.ApsClient(FakeAuth(), transport=router.transport)
|
||||
|
||||
|
||||
# --- pure helpers ------------------------------------------------------------
|
||||
|
||||
|
||||
def test_base64url_strips_padding():
|
||||
assert autodesk._base64url(b"\x00") == "AA"
|
||||
assert "=" not in autodesk._base64url(b"\x00\x00")
|
||||
|
||||
|
||||
def test_generate_code_verifier_is_url_safe_and_unique():
|
||||
verifier = autodesk.generate_code_verifier()
|
||||
assert not set(verifier) & set("=+/")
|
||||
assert autodesk.generate_code_verifier() != autodesk.generate_code_verifier()
|
||||
|
||||
|
||||
def test_generate_code_challenge_matches_rfc7636_vector():
|
||||
# RFC 7636 Appendix B test vector.
|
||||
verifier = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
|
||||
assert (
|
||||
autodesk.generate_code_challenge(verifier)
|
||||
== "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
|
||||
)
|
||||
|
||||
|
||||
def test_parse_storage_id_splits_bucket_and_object():
|
||||
bucket, obj = autodesk.ApsClient._parse_storage_id(
|
||||
"urn:adsk.objects:os.object:wip.dm.prod/abc-123.ifc"
|
||||
)
|
||||
assert bucket == "wip.dm.prod"
|
||||
assert obj == "abc-123.ifc"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bad",
|
||||
[
|
||||
"not-a-urn",
|
||||
"urn:adsk.objects:os.object:bucketonly",
|
||||
"urn:adsk.objects:os.object:/object",
|
||||
"urn:adsk.objects:os.object:bucket/",
|
||||
],
|
||||
)
|
||||
def test_parse_storage_id_rejects_malformed(bad):
|
||||
with pytest.raises(RpcError):
|
||||
autodesk.ApsClient._parse_storage_id(bad)
|
||||
|
||||
|
||||
def test_entry_extracts_fields():
|
||||
item = {
|
||||
"id": "x",
|
||||
"type": "items",
|
||||
"attributes": {
|
||||
"displayName": "Model.ifc",
|
||||
"extension": {"type": "items:autodesk.bim360:File"},
|
||||
},
|
||||
}
|
||||
assert autodesk.ApsClient._entry(item) == {
|
||||
"id": "x",
|
||||
"type": "items",
|
||||
"display_name": "Model.ifc",
|
||||
"name": None,
|
||||
"extension_type": "items:autodesk.bim360:File",
|
||||
}
|
||||
|
||||
|
||||
def test_relationship_id_handles_dict_list_and_missing():
|
||||
data = {
|
||||
"relationships": {
|
||||
"parent": {"data": {"id": "p1"}},
|
||||
"files": {"data": [{"id": "f1"}, {"id": "f2"}]},
|
||||
}
|
||||
}
|
||||
assert autodesk.ApsClient._relationship_id(data, "parent") == "p1"
|
||||
assert autodesk.ApsClient._relationship_id(data, "files") == "f1"
|
||||
assert autodesk.ApsClient._relationship_id(data, "missing") is None
|
||||
|
||||
|
||||
def test_entry_name_matches_is_case_insensitive():
|
||||
assert autodesk.ApsClient._entry_name_matches({"display_name": "Model.IFC"}, "model.ifc")
|
||||
assert autodesk.ApsClient._entry_name_matches({"name": "Model.IFC"}, "model.ifc")
|
||||
assert not autodesk.ApsClient._entry_name_matches({"display_name": "other.ifc"}, "model.ifc")
|
||||
|
||||
|
||||
# --- KeyringTokenStore -------------------------------------------------------
|
||||
|
||||
|
||||
def test_keyring_token_store_round_trip(memory_keyring):
|
||||
store = autodesk.KeyringTokenStore(service_name="svc", username="user")
|
||||
assert store.load() is None
|
||||
store.save({"access_token": "abc"})
|
||||
assert store.load() == {"access_token": "abc"}
|
||||
store.delete()
|
||||
assert store.load() is None
|
||||
|
||||
|
||||
def test_keyring_token_store_delete_missing_is_noop(memory_keyring):
|
||||
store = autodesk.KeyringTokenStore(service_name="svc", username="user")
|
||||
store.delete() # no entry to delete — must not raise
|
||||
|
||||
|
||||
def test_keyring_missing_backend_surfaces_friendly_error(monkeypatch):
|
||||
def boom(*_args, **_kwargs):
|
||||
raise keyring.errors.NoKeyringError("no backend")
|
||||
|
||||
monkeypatch.setattr(autodesk.keyring, "get_password", boom)
|
||||
store = autodesk.KeyringTokenStore(service_name="svc", username="user")
|
||||
with pytest.raises(RpcError) as excinfo:
|
||||
store.load()
|
||||
assert "keyring" in excinfo.value.message.lower()
|
||||
|
||||
|
||||
# --- token lifecycle / injected clock ---------------------------------------
|
||||
|
||||
|
||||
def test_ensure_access_token_returns_unexpired_token_without_http():
|
||||
store = FakeTokenStore(stored_token_dict(access_offset=3600, refresh_offset=100_000))
|
||||
router = Router()
|
||||
auth = make_auth(router=router, token_store=store)
|
||||
assert auth.ensure_access_token() == "current-access"
|
||||
assert router.requests == []
|
||||
|
||||
|
||||
def test_ensure_access_token_refreshes_when_access_expired():
|
||||
store = FakeTokenStore(stored_token_dict(access_offset=-100, refresh_offset=100_000))
|
||||
router = Router().add(
|
||||
"POST",
|
||||
"/authentication/v2/token",
|
||||
{
|
||||
"json": {
|
||||
"access_token": "refreshed-access",
|
||||
"refresh_token": "refreshed-refresh",
|
||||
"expires_in": 3600,
|
||||
"refresh_token_expires_in": 200_000,
|
||||
}
|
||||
},
|
||||
)
|
||||
auth = make_auth(router=router, token_store=store)
|
||||
assert auth.ensure_access_token() == "refreshed-access"
|
||||
assert store.value["access_token"] == "refreshed-access"
|
||||
assert router.count("POST", "/authentication/v2/token") == 1
|
||||
|
||||
|
||||
def test_ensure_access_token_logs_in_when_both_tokens_expired(monkeypatch):
|
||||
monkeypatch.setattr(autodesk.webbrowser, "open", lambda _url: None)
|
||||
store = FakeTokenStore() # nothing stored at all
|
||||
router = Router().add(
|
||||
"POST",
|
||||
"/authentication/v2/token",
|
||||
{
|
||||
"json": {
|
||||
"access_token": "logged-in-access",
|
||||
"refresh_token": "logged-in-refresh",
|
||||
"expires_in": 3600,
|
||||
}
|
||||
},
|
||||
)
|
||||
auth = make_auth(router=router, token_store=store, callback_waiter=lambda *_a: "auth-code")
|
||||
assert auth.ensure_access_token() == "logged-in-access"
|
||||
assert store.value["access_token"] == "logged-in-access"
|
||||
|
||||
|
||||
def test_token_from_payload_uses_injected_clock():
|
||||
auth = make_auth()
|
||||
token = auth._token_from_payload(
|
||||
{"access_token": "a", "refresh_token": "r", "expires_in": 3600}
|
||||
)
|
||||
assert token.access_token_expires_at_utc == iso(3600 - 30)
|
||||
# Default refresh TTL is 15 days, minus the same 30s safety margin.
|
||||
assert token.refresh_token_expires_at_utc == iso(15 * 24 * 60 * 60 - 30)
|
||||
|
||||
|
||||
def test_login_interactive_rejects_non_local_callback():
|
||||
auth = make_auth(callback_url="https://example.com/callback")
|
||||
with pytest.raises(RpcError, match="Callback URL"):
|
||||
auth.login_interactive()
|
||||
|
||||
|
||||
def test_login_interactive_exchanges_code_for_token(monkeypatch):
|
||||
opened: list[str] = []
|
||||
monkeypatch.setattr(autodesk.webbrowser, "open", lambda url: opened.append(url))
|
||||
store = FakeTokenStore()
|
||||
router = Router().add(
|
||||
"POST",
|
||||
"/authentication/v2/token",
|
||||
{
|
||||
"json": {
|
||||
"access_token": "fresh-access",
|
||||
"refresh_token": "fresh-refresh",
|
||||
"expires_in": 3600,
|
||||
}
|
||||
},
|
||||
)
|
||||
auth = make_auth(router=router, token_store=store, callback_waiter=lambda *_a: "the-code")
|
||||
token = auth.login_interactive()
|
||||
assert token.access_token == "fresh-access"
|
||||
assert store.value["access_token"] == "fresh-access"
|
||||
assert opened and opened[0].startswith(autodesk.AuthSessionService.authorize_endpoint)
|
||||
|
||||
|
||||
def test_token_exchange_failure_raises_rpc_error(monkeypatch):
|
||||
monkeypatch.setattr(autodesk.webbrowser, "open", lambda _url: None)
|
||||
router = Router().add(
|
||||
"POST", "/authentication/v2/token", {"status": 400, "text": "invalid_grant"}
|
||||
)
|
||||
auth = make_auth(router=router, callback_waiter=lambda *_a: "the-code")
|
||||
with pytest.raises(RpcError, match="invalid_grant"):
|
||||
auth.login_interactive()
|
||||
|
||||
|
||||
# --- wait_for_oauth_callback (real loopback socket) --------------------------
|
||||
|
||||
|
||||
def _free_port() -> int:
|
||||
with socket.socket() as probe:
|
||||
probe.bind(("127.0.0.1", 0))
|
||||
return probe.getsockname()[1]
|
||||
|
||||
|
||||
def _get_with_retry(url: str, params: dict, timeout: float = 5.0) -> None:
|
||||
"""Fire one GET, retrying only while the server has not yet bound."""
|
||||
deadline = time.time() + timeout
|
||||
while True:
|
||||
try:
|
||||
httpx.get(url, params=params)
|
||||
return
|
||||
except httpx.ConnectError:
|
||||
if time.time() > deadline:
|
||||
raise
|
||||
time.sleep(0.02)
|
||||
|
||||
|
||||
def drive_callback(expected_state: str, query: dict, path: str = "/cb") -> dict:
|
||||
"""Run ``wait_for_oauth_callback`` in a thread and fire one redirect at it."""
|
||||
port = _free_port()
|
||||
outcome: dict = {}
|
||||
|
||||
def server() -> None:
|
||||
try:
|
||||
outcome["code"] = autodesk.wait_for_oauth_callback(
|
||||
"127.0.0.1", port, path, expected_state
|
||||
)
|
||||
except BaseException as exc: # noqa: BLE001 - re-raised to the test
|
||||
outcome["error"] = exc
|
||||
|
||||
thread = threading.Thread(target=server, daemon=True)
|
||||
thread.start()
|
||||
_get_with_retry(f"http://127.0.0.1:{port}{path}", query)
|
||||
thread.join(timeout=5)
|
||||
return outcome
|
||||
|
||||
|
||||
def test_wait_for_oauth_callback_returns_authorization_code():
|
||||
outcome = drive_callback("state-123", {"state": "state-123", "code": "the-code"})
|
||||
assert outcome.get("code") == "the-code"
|
||||
|
||||
|
||||
def test_wait_for_oauth_callback_rejects_state_mismatch():
|
||||
outcome = drive_callback("expected-state", {"state": "tampered", "code": "c"})
|
||||
assert isinstance(outcome.get("error"), RpcError)
|
||||
assert "state mismatch" in outcome["error"].message.lower()
|
||||
|
||||
|
||||
def test_wait_for_oauth_callback_reports_oauth_error():
|
||||
outcome = drive_callback("state-123", {"state": "state-123", "error": "access_denied"})
|
||||
assert isinstance(outcome.get("error"), RpcError)
|
||||
assert "access_denied" in outcome["error"].message
|
||||
|
||||
|
||||
def test_wait_for_oauth_callback_requires_a_code():
|
||||
outcome = drive_callback("state-123", {"state": "state-123"})
|
||||
assert isinstance(outcome.get("error"), RpcError)
|
||||
assert "authorization code" in outcome["error"].message.lower()
|
||||
|
||||
|
||||
# --- ApsClient browsing ------------------------------------------------------
|
||||
|
||||
|
||||
def _hub(hub_id: str, name: str) -> dict:
|
||||
return {
|
||||
"id": hub_id,
|
||||
"attributes": {"name": name, "extension": {"type": "hubs:autodesk.core:Hub"}},
|
||||
}
|
||||
|
||||
|
||||
def _project(project_id: str, name: str) -> dict:
|
||||
return {
|
||||
"id": project_id,
|
||||
"attributes": {
|
||||
"name": name,
|
||||
"extension": {"type": "projects:autodesk.bim360:Project"},
|
||||
},
|
||||
"relationships": {"rootFolder": {"data": {"id": f"root-{project_id}"}}},
|
||||
}
|
||||
|
||||
|
||||
def test_list_hubs_sorts_case_insensitively():
|
||||
router = Router().add(
|
||||
"GET",
|
||||
"/project/v1/hubs",
|
||||
{"json": {"data": [_hub("h2", "Beta"), _hub("h1", "alpha")]}},
|
||||
)
|
||||
hubs = make_client(router).list_hubs()
|
||||
assert [h["name"] for h in hubs] == ["alpha", "Beta"]
|
||||
assert hubs[0]["id"] == "h1"
|
||||
|
||||
|
||||
def test_list_projects_follows_pagination():
|
||||
router = Router()
|
||||
router.add(
|
||||
"GET",
|
||||
"page=2",
|
||||
{"json": {"data": [_project("p2", "Zeta")], "links": {}}},
|
||||
)
|
||||
router.add(
|
||||
"GET",
|
||||
"/hubs/h/projects",
|
||||
{
|
||||
"json": {
|
||||
"data": [_project("p1", "Alpha")],
|
||||
"links": {
|
||||
"next": {
|
||||
"href": "https://developer.api.autodesk.com/project/v1/hubs/h/projects?page=2"
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
projects = make_client(router).list_projects("h")
|
||||
assert [p["id"] for p in projects] == ["p1", "p2"]
|
||||
assert projects[0]["root_folder_id"] == "root-p1"
|
||||
|
||||
|
||||
def test_get_item_returns_tip_details():
|
||||
router = Router().add(
|
||||
"GET",
|
||||
"/items/",
|
||||
{
|
||||
"json": {
|
||||
"data": {
|
||||
"id": "item-1",
|
||||
"attributes": {"displayName": "Model.ifcfed", "hidden": False},
|
||||
"relationships": {
|
||||
"parent": {"data": {"id": "folder-1"}},
|
||||
"tip": {"data": {"id": "v3"}},
|
||||
},
|
||||
},
|
||||
"included": [
|
||||
{
|
||||
"type": "versions",
|
||||
"id": "v3",
|
||||
"attributes": {
|
||||
"versionNumber": 3,
|
||||
"lastModifiedTime": "2026-01-01T00:00:00Z",
|
||||
"lastModifiedUserName": "Dion",
|
||||
},
|
||||
"relationships": {
|
||||
"storage": {
|
||||
"data": {"id": "urn:adsk.objects:os.object:b/o"}
|
||||
}
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
},
|
||||
)
|
||||
item = make_client(router).get_item("proj-1", "item-1")
|
||||
assert item["hidden"] is False
|
||||
assert item["version_id"] == "v3"
|
||||
assert item["storage_id"] == "urn:adsk.objects:os.object:b/o"
|
||||
assert item["version_number"] == 3
|
||||
assert item["parent_folder_id"] == "folder-1"
|
||||
assert item["last_modified_user_name"] == "Dion"
|
||||
|
||||
|
||||
def test_get_item_without_tip_is_treated_as_hidden():
|
||||
router = Router().add(
|
||||
"GET",
|
||||
"/items/",
|
||||
{"json": {"data": {"id": "item-1", "attributes": {}, "relationships": {}}}},
|
||||
)
|
||||
item = make_client(router).get_item("proj-1", "item-1")
|
||||
assert item["hidden"] is True
|
||||
assert item["storage_id"] is None
|
||||
assert item["version_id"] is None
|
||||
|
||||
|
||||
def test_list_folder_contents_applies_extension_filter():
|
||||
router = Router().add(
|
||||
"GET",
|
||||
"/contents",
|
||||
{
|
||||
"json": {
|
||||
"data": [
|
||||
{
|
||||
"id": "f1",
|
||||
"type": "folders",
|
||||
"attributes": {"name": "Sub", "extension": {"type": "t"}},
|
||||
},
|
||||
{
|
||||
"id": "i1",
|
||||
"type": "items",
|
||||
"attributes": {"displayName": "keep.ifcfed", "extension": {}},
|
||||
},
|
||||
{
|
||||
"id": "i2",
|
||||
"type": "items",
|
||||
"attributes": {"displayName": "skip.txt", "extension": {}},
|
||||
},
|
||||
],
|
||||
"links": {},
|
||||
}
|
||||
},
|
||||
)
|
||||
entries = make_client(router).list_folder_contents(
|
||||
"proj",
|
||||
"folder",
|
||||
extension_filter=lambda e: (e["display_name"] or "").endswith(".ifcfed"),
|
||||
)
|
||||
ids = {e["id"] for e in entries}
|
||||
assert ids == {"f1", "i1"} # folders kept, non-.ifcfed item dropped
|
||||
|
||||
|
||||
def test_get_json_maps_http_error_to_rpc_error():
|
||||
router = Router().add(
|
||||
"GET", "/project/v1/hubs", {"status": 403, "text": "Forbidden: bad token"}
|
||||
)
|
||||
with pytest.raises(RpcError, match="Forbidden"):
|
||||
make_client(router).list_hubs()
|
||||
|
||||
|
||||
# --- download / upload -------------------------------------------------------
|
||||
|
||||
|
||||
def test_download_storage_to_file_writes_content_and_reports_progress(tmp_path):
|
||||
router = Router()
|
||||
router.add(
|
||||
"GET",
|
||||
"/signeds3download",
|
||||
{"json": {"url": "https://signed.example/blob"}},
|
||||
)
|
||||
router.add("GET", "signed.example/blob", {"content": b"hello world"})
|
||||
dest = tmp_path / "out.bin"
|
||||
seen: list = []
|
||||
make_client(router).download_storage_to_file(
|
||||
"urn:adsk.objects:os.object:bucket/object",
|
||||
dest,
|
||||
progress=lambda name, pct, done, total: seen.append((name, pct, done, total)),
|
||||
)
|
||||
assert dest.read_bytes() == b"hello world"
|
||||
assert seen[-1][1] == 100 # final progress callback reports 100%
|
||||
|
||||
|
||||
def test_signed_download_url_missing_raises():
|
||||
router = Router().add("GET", "/signeds3download", {"json": {}})
|
||||
with pytest.raises(RpcError, match="URL"):
|
||||
make_client(router).download_storage_to_file(
|
||||
"urn:adsk.objects:os.object:bucket/object", "/tmp/ignored"
|
||||
)
|
||||
|
||||
|
||||
def _upload_router() -> Router:
|
||||
router = Router()
|
||||
router.add(
|
||||
"POST", "/storage", {"json": {"data": {"id": "urn:adsk.objects:os.object:bk/obj"}}}
|
||||
)
|
||||
router.add(
|
||||
"GET",
|
||||
"/signeds3upload",
|
||||
{"json": {"uploadKey": "ukey", "urls": ["https://up.example/part1"]}},
|
||||
)
|
||||
router.add("PUT", "up.example/part1", {"status": 200})
|
||||
router.add("POST", "/signeds3upload", {"status": 200, "json": {}})
|
||||
return router
|
||||
|
||||
|
||||
def test_upload_file_creates_new_item_when_folder_is_empty(tmp_path):
|
||||
local = tmp_path / "model.ifc"
|
||||
local.write_bytes(b"x" * 1024)
|
||||
router = _upload_router()
|
||||
router.add("GET", "/contents", {"json": {"data": [], "links": {}}})
|
||||
router.add(
|
||||
"POST",
|
||||
"/items",
|
||||
{
|
||||
"json": {
|
||||
"data": {"id": "new-item"},
|
||||
"included": [
|
||||
{
|
||||
"type": "versions",
|
||||
"id": "v1",
|
||||
"attributes": {
|
||||
"versionNumber": 1,
|
||||
"lastModifiedTime": "2026-01-01T00:00:00Z",
|
||||
"lastModifiedUserName": "Dion",
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
},
|
||||
)
|
||||
result = make_client(router).upload_file_to_folder(
|
||||
"proj", "folder", local, display_name="model.ifc"
|
||||
)
|
||||
assert result["item_id"] == "new-item"
|
||||
assert result["version_id"] == "v1"
|
||||
assert result["version_number"] == 1
|
||||
assert router.count("PUT", "up.example/part1") == 1
|
||||
|
||||
|
||||
def test_upload_file_creates_a_version_when_item_exists(tmp_path):
|
||||
local = tmp_path / "model.ifc"
|
||||
local.write_bytes(b"x" * 1024)
|
||||
router = _upload_router()
|
||||
router.add(
|
||||
"GET",
|
||||
"/contents",
|
||||
{
|
||||
"json": {
|
||||
"data": [
|
||||
{
|
||||
"id": "existing-item",
|
||||
"type": "items",
|
||||
"attributes": {"displayName": "model.ifc", "extension": {}},
|
||||
}
|
||||
],
|
||||
"links": {},
|
||||
}
|
||||
},
|
||||
)
|
||||
router.add(
|
||||
"POST",
|
||||
"/versions",
|
||||
{"json": {"data": {"id": "v7", "attributes": {"versionNumber": 7}}}},
|
||||
)
|
||||
result = make_client(router).upload_file_to_folder(
|
||||
"proj", "folder", local, display_name="model.ifc"
|
||||
)
|
||||
assert result["item_id"] == "existing-item"
|
||||
assert result["version_id"] == "v7"
|
||||
assert result["version_number"] == 7
|
||||
|
||||
|
||||
def test_upload_rejects_missing_local_file(tmp_path):
|
||||
with pytest.raises(RpcError, match="does not exist"):
|
||||
make_client(Router()).upload_file_to_folder(
|
||||
"proj", "folder", tmp_path / "missing.ifc"
|
||||
)
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Tests for the on-disk cache layout (``bonsaiviewer_autodesk.cache``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bonsaiviewer_autodesk import cache
|
||||
|
||||
|
||||
def test_cache_root_uses_xdg_on_linux(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(cache.platform, "system", lambda: "Linux")
|
||||
monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "xdg"))
|
||||
root = cache.cache_root()
|
||||
assert root == tmp_path / "xdg" / "bonsaiviewer-autodesk"
|
||||
assert root.is_dir()
|
||||
|
||||
|
||||
def test_cache_root_uses_localappdata_on_windows(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(cache.platform, "system", lambda: "Windows")
|
||||
monkeypatch.setenv("LOCALAPPDATA", str(tmp_path / "appdata"))
|
||||
root = cache.cache_root()
|
||||
assert root == tmp_path / "appdata" / "bonsaiviewer-autodesk" / "Cache"
|
||||
assert root.is_dir()
|
||||
|
||||
|
||||
def test_cache_root_uses_library_caches_on_macos(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(cache.platform, "system", lambda: "Darwin")
|
||||
monkeypatch.setattr(cache.Path, "home", lambda: tmp_path / "home")
|
||||
root = cache.cache_root()
|
||||
assert root == tmp_path / "home" / "Library" / "Caches" / "bonsaiviewer-autodesk"
|
||||
assert root.is_dir()
|
||||
|
||||
|
||||
def test_ifcfed_dir_is_deterministic(cache_dir):
|
||||
first = cache.ifcfed_dir("project-1", "item-1")
|
||||
second = cache.ifcfed_dir("project-1", "item-1")
|
||||
assert first == second
|
||||
|
||||
|
||||
def test_ifcfed_dir_varies_with_inputs(cache_dir):
|
||||
assert cache.ifcfed_dir("p", "item-1") != cache.ifcfed_dir("p", "item-2")
|
||||
assert cache.ifcfed_dir("p1", "item") != cache.ifcfed_dir("p2", "item")
|
||||
|
||||
|
||||
def test_model_dir_is_per_version(cache_dir):
|
||||
v1 = cache.model_dir("p", "item", "v1")
|
||||
v2 = cache.model_dir("p", "item", "v2")
|
||||
assert v1 != v2
|
||||
assert cache.model_dir("p", "item", "v1") == v1
|
||||
|
||||
|
||||
def test_prepare_sole_child_dir_wipes_existing_contents(cache_dir):
|
||||
directory = cache.ifcfed_dir("p", "item")
|
||||
directory.mkdir(parents=True)
|
||||
(directory / "stale.txt").write_text("old")
|
||||
|
||||
result = cache.prepare_sole_child_dir(directory)
|
||||
|
||||
assert result == directory
|
||||
assert directory.is_dir()
|
||||
assert list(directory.iterdir()) == []
|
||||
|
||||
|
||||
def test_prepare_sole_child_dir_creates_when_absent(cache_dir):
|
||||
directory = cache.model_dir("p", "item", "v1")
|
||||
assert not directory.exists()
|
||||
cache.prepare_sole_child_dir(directory)
|
||||
assert directory.is_dir()
|
||||
|
||||
|
||||
def test_manifest_round_trips(cache_dir):
|
||||
directory = cache.prepare_sole_child_dir(cache.ifcfed_dir("p", "item"))
|
||||
ifcfed_path = directory / "model.ifcfed"
|
||||
ifcfed_path.write_text("data")
|
||||
manifest = {"connector": "autodesk", "item_id": "item", "hub_id": "h"}
|
||||
|
||||
manifest_path = cache.write_manifest(ifcfed_path, manifest)
|
||||
|
||||
assert manifest_path.name == "model.ifcfed.manifest"
|
||||
assert cache.read_manifest(ifcfed_path) == manifest
|
||||
|
||||
|
||||
def test_read_manifest_returns_none_when_absent(cache_dir):
|
||||
directory = cache.prepare_sole_child_dir(cache.ifcfed_dir("p", "item"))
|
||||
assert cache.read_manifest(directory / "model.ifcfed") is None
|
||||
@@ -0,0 +1,305 @@
|
||||
"""Tests for the RPC handlers (``bonsaiviewer_autodesk.connector``).
|
||||
|
||||
The non-interactive handlers are exercised against a fake ``ApsClient`` so no
|
||||
network or GUI is touched; ``progress_dialog`` is stubbed out.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from bonsaiviewer_autodesk import cache, connector, settings
|
||||
from bonsaiviewer_autodesk.rpc import RpcError
|
||||
|
||||
|
||||
# --- fakes / fixtures --------------------------------------------------------
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _fake_progress(_message):
|
||||
yield lambda *_args, **_kwargs: None
|
||||
|
||||
|
||||
class FakeAps:
|
||||
"""Stand-in for ApsClient covering only what the handlers call."""
|
||||
|
||||
def __init__(self, *, item: dict | None = None, items: dict | None = None) -> None:
|
||||
self._item = item
|
||||
self._items = items or {}
|
||||
self.downloaded: list[tuple] = []
|
||||
self.uploaded: list[tuple] = []
|
||||
|
||||
def get_item(self, _project_id: str, item_id: str) -> dict:
|
||||
if item_id in self._items:
|
||||
return dict(self._items[item_id])
|
||||
assert self._item is not None, f"no canned item for {item_id!r}"
|
||||
return dict(self._item)
|
||||
|
||||
def download_storage_to_file(self, storage_id, destination_path, *, progress=None) -> None:
|
||||
Path(destination_path).write_bytes(b"<ifc data>")
|
||||
self.downloaded.append((storage_id, Path(destination_path)))
|
||||
|
||||
def upload_file_to_folder(
|
||||
self, project_id, folder_id, local_path, *, display_name=None, progress=None
|
||||
) -> dict:
|
||||
self.uploaded.append((project_id, folder_id, Path(local_path), display_name))
|
||||
return {
|
||||
"item_id": "uploaded-item",
|
||||
"version_id": "v1",
|
||||
"version_number": 1,
|
||||
"last_modified_time_utc": "2026-01-01T00:00:00Z",
|
||||
"last_modified_user_name": "Dion",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def make_connector(config_dir, cache_dir, monkeypatch):
|
||||
"""Build an AutodeskConnector wired to a fake ApsClient."""
|
||||
monkeypatch.setattr(connector, "progress_dialog", _fake_progress)
|
||||
|
||||
def _make(aps: FakeAps) -> connector.AutodeskConnector:
|
||||
conn = connector.AutodeskConnector()
|
||||
conn.aps = aps
|
||||
conn.auth = object() # only identity matters to _require_aps
|
||||
return conn
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
def _ifcfed_item(**overrides) -> dict:
|
||||
item = {
|
||||
"id": "item-1",
|
||||
"hidden": False,
|
||||
"storage_id": "urn:adsk.objects:os.object:bucket/object",
|
||||
"display_name": "Project.ifcfed",
|
||||
"version_id": "v1",
|
||||
"version_number": 1,
|
||||
"last_modified_time_utc": None,
|
||||
"last_modified_user_name": None,
|
||||
"parent_folder_id": "folder-1",
|
||||
}
|
||||
item.update(overrides)
|
||||
return item
|
||||
|
||||
|
||||
# --- pure helpers ------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,expected",
|
||||
[
|
||||
(0, "0 B"),
|
||||
(512, "512 B"),
|
||||
(1024, "1.0 KB"),
|
||||
(1536, "1.5 KB"),
|
||||
(5 * 1024 * 1024, "5.0 MB"),
|
||||
],
|
||||
)
|
||||
def test_format_bytes(value, expected):
|
||||
assert connector._format_bytes(value) == expected
|
||||
|
||||
|
||||
def test_progress_detail_combines_percent_and_bytes():
|
||||
detail = connector._progress_detail(45, 4_500_000, 10_000_000)
|
||||
assert detail.startswith("45%, ")
|
||||
assert " / " in detail
|
||||
|
||||
|
||||
def test_progress_detail_percent_only():
|
||||
assert connector._progress_detail(50, None, None) == "50%"
|
||||
|
||||
|
||||
def test_progress_detail_bytes_without_total():
|
||||
assert connector._progress_detail(None, 2048, None) == "2.0 KB"
|
||||
|
||||
|
||||
def test_progress_detail_empty_when_nothing_known():
|
||||
assert connector._progress_detail(None, None, None) == ""
|
||||
|
||||
|
||||
def test_build_metadata_maps_all_fields():
|
||||
metadata = connector._build_metadata(
|
||||
{
|
||||
"version_number": 3,
|
||||
"last_modified_time_utc": "2026-01-01T00:00:00Z",
|
||||
"last_modified_user_name": "Dion",
|
||||
}
|
||||
)
|
||||
assert metadata == {
|
||||
"revision": "v3",
|
||||
"date": "2026-01-01T00:00:00Z",
|
||||
"author": "Dion",
|
||||
}
|
||||
|
||||
|
||||
def test_build_metadata_omits_missing_fields():
|
||||
assert connector._build_metadata({}) == {}
|
||||
assert connector._build_metadata({"version_number": None}) == {}
|
||||
|
||||
|
||||
def test_require_string_rejects_missing_or_blank():
|
||||
assert connector._require_string({"k": "v"}, "k") == "v"
|
||||
with pytest.raises(RpcError):
|
||||
connector._require_string({"k": " "}, "k")
|
||||
with pytest.raises(RpcError):
|
||||
connector._require_string({}, "k")
|
||||
|
||||
|
||||
def test_require_object_and_array_type_checks():
|
||||
assert connector._require_object({"a": 1}, "p") == {"a": 1}
|
||||
assert connector._require_array([1, 2], "p") == [1, 2]
|
||||
with pytest.raises(RpcError):
|
||||
connector._require_object([], "p")
|
||||
with pytest.raises(RpcError):
|
||||
connector._require_array({}, "p")
|
||||
|
||||
|
||||
# --- credential wiring -------------------------------------------------------
|
||||
|
||||
|
||||
def test_reload_credentials_without_client_id_leaves_aps_unset(config_dir):
|
||||
conn = connector.AutodeskConnector()
|
||||
assert conn.aps is None
|
||||
assert conn.auth is None
|
||||
with pytest.raises(RpcError, match="client id"):
|
||||
conn._require_aps()
|
||||
|
||||
|
||||
def test_reload_credentials_with_client_id_builds_aps(config_dir):
|
||||
settings.save_client_id("my-client-id")
|
||||
conn = connector.AutodeskConnector()
|
||||
assert conn.aps is not None
|
||||
assert conn.auth is not None
|
||||
|
||||
|
||||
# --- pull_ifcfed -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_pull_ifcfed_downloads_and_writes_manifest(make_connector):
|
||||
aps = FakeAps(item=_ifcfed_item())
|
||||
conn = make_connector(aps)
|
||||
|
||||
result = conn.pull_ifcfed({"hub_id": "h", "project_id": "p", "item_id": "item-1"})
|
||||
|
||||
path = Path(result["path"])
|
||||
assert path.exists()
|
||||
assert path.name == "Project.ifcfed"
|
||||
assert aps.downloaded # the fake actually got asked to download
|
||||
|
||||
manifest = cache.read_manifest(path)
|
||||
assert manifest["connector"] == "autodesk"
|
||||
assert manifest["item_id"] == "item-1"
|
||||
assert manifest["hub_id"] == "h"
|
||||
|
||||
|
||||
def test_pull_ifcfed_rejects_non_ifcfed_file(make_connector):
|
||||
aps = FakeAps(item=_ifcfed_item(display_name="model.ifc"))
|
||||
conn = make_connector(aps)
|
||||
with pytest.raises(RpcError, match="ifcfed"):
|
||||
conn.pull_ifcfed({"hub_id": "h", "project_id": "p", "item_id": "item-1"})
|
||||
|
||||
|
||||
def test_pull_ifcfed_rejects_deleted_item(make_connector):
|
||||
aps = FakeAps(item=_ifcfed_item(hidden=True))
|
||||
conn = make_connector(aps)
|
||||
with pytest.raises(RpcError, match="deleted"):
|
||||
conn.pull_ifcfed({"hub_id": "h", "project_id": "p", "item_id": "item-1"})
|
||||
|
||||
|
||||
def test_pull_ifcfed_requires_string_fields(make_connector):
|
||||
conn = make_connector(FakeAps(item=_ifcfed_item()))
|
||||
with pytest.raises(RpcError, match="item_id"):
|
||||
conn.pull_ifcfed({"hub_id": "h", "project_id": "p"})
|
||||
|
||||
|
||||
# --- pull_models -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_pull_models_skips_failures_with_none(make_connector):
|
||||
aps = FakeAps(
|
||||
items={
|
||||
"good": _ifcfed_item(id="good", display_name="good.ifc"),
|
||||
"gone": _ifcfed_item(id="gone", hidden=True),
|
||||
}
|
||||
)
|
||||
conn = make_connector(aps)
|
||||
models = [
|
||||
{"source": {"connector": "autodesk", "project_id": "p", "item_id": "good"}},
|
||||
{"source": {"connector": "autodesk", "project_id": "p", "item_id": "gone"}},
|
||||
{"source": {"connector": "other", "project_id": "p", "item_id": "good"}},
|
||||
]
|
||||
|
||||
results = conn.pull_models(models)
|
||||
|
||||
assert len(results) == 3
|
||||
assert results[0] is not None and Path(results[0]["path"]).exists()
|
||||
assert results[1] is None # hidden/deleted item
|
||||
assert results[2] is None # wrong connector -> RpcError, swallowed
|
||||
|
||||
|
||||
def test_pull_models_requires_a_json_array(make_connector):
|
||||
conn = make_connector(FakeAps(item=_ifcfed_item()))
|
||||
with pytest.raises(RpcError, match="array"):
|
||||
conn.pull_models({"not": "an array"})
|
||||
|
||||
|
||||
# --- push_ifcfed -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_push_ifcfed_uploads_and_caches_with_manifest(make_connector, tmp_path):
|
||||
local = tmp_path / "local.ifcfed"
|
||||
local.write_bytes(b"ifcfed bytes")
|
||||
aps = FakeAps(item=_ifcfed_item())
|
||||
conn = make_connector(aps)
|
||||
|
||||
result = conn.push_ifcfed(
|
||||
{
|
||||
"path": str(local),
|
||||
"manifest": {
|
||||
"connector": "autodesk",
|
||||
"hub_id": "h",
|
||||
"project_id": "p",
|
||||
"item_id": "item-1",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
assert aps.uploaded
|
||||
cached = Path(result["path"])
|
||||
assert cached.exists()
|
||||
assert cache.read_manifest(cached)["item_id"] == "uploaded-item"
|
||||
|
||||
|
||||
def test_push_ifcfed_rejects_missing_local_file(make_connector):
|
||||
conn = make_connector(FakeAps(item=_ifcfed_item()))
|
||||
with pytest.raises(RpcError, match="does not exist"):
|
||||
conn.push_ifcfed(
|
||||
{
|
||||
"path": "/no/such/file.ifcfed",
|
||||
"manifest": {
|
||||
"connector": "autodesk",
|
||||
"hub_id": "h",
|
||||
"project_id": "p",
|
||||
"item_id": "item-1",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_push_ifcfed_rejects_non_ifcfed_extension(make_connector, tmp_path):
|
||||
local = tmp_path / "local.ifc"
|
||||
local.write_bytes(b"data")
|
||||
conn = make_connector(FakeAps(item=_ifcfed_item()))
|
||||
with pytest.raises(RpcError, match="ifcfed"):
|
||||
conn.push_ifcfed({"path": str(local), "manifest": {"connector": "autodesk"}})
|
||||
|
||||
|
||||
def test_push_ifcfed_rejects_foreign_connector_manifest(make_connector, tmp_path):
|
||||
local = tmp_path / "local.ifcfed"
|
||||
local.write_bytes(b"data")
|
||||
conn = make_connector(FakeAps(item=_ifcfed_item()))
|
||||
with pytest.raises(RpcError, match="connector"):
|
||||
conn.push_ifcfed({"path": str(local), "manifest": {"connector": "other"}})
|
||||
@@ -0,0 +1,123 @@
|
||||
"""Tests for the JSON-RPC host (``bonsaiviewer_autodesk.rpc``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
|
||||
from bonsaiviewer_autodesk.rpc import (
|
||||
JSONRPC_INTERNAL_ERROR,
|
||||
JSONRPC_INVALID_PARAMS,
|
||||
JSONRPC_INVALID_REQUEST,
|
||||
JSONRPC_METHOD_NOT_FOUND,
|
||||
JSONRPC_PARSE_ERROR,
|
||||
JsonRpcHost,
|
||||
RpcError,
|
||||
)
|
||||
|
||||
|
||||
def run(line: str, handlers: dict | None = None) -> tuple[list[dict], str]:
|
||||
"""Drive a ``JsonRpcHost`` over in-memory streams.
|
||||
|
||||
Returns ``(responses, stderr_text)`` where ``responses`` is the parsed
|
||||
JSON written to stdout, one element per line.
|
||||
"""
|
||||
out = io.StringIO()
|
||||
err = io.StringIO()
|
||||
JsonRpcHost(
|
||||
handlers or {},
|
||||
stdin=io.StringIO(line),
|
||||
stdout=out,
|
||||
stderr=err,
|
||||
).run()
|
||||
responses = [json.loads(piece) for piece in out.getvalue().splitlines() if piece]
|
||||
return responses, err.getvalue()
|
||||
|
||||
|
||||
def test_parse_error_for_invalid_json():
|
||||
responses, _ = run("this is not json\n")
|
||||
assert responses[0]["error"]["code"] == JSONRPC_PARSE_ERROR
|
||||
assert responses[0]["id"] is None
|
||||
|
||||
|
||||
def test_request_must_be_a_json_object():
|
||||
responses, _ = run("[1, 2, 3]\n")
|
||||
assert responses[0]["error"]["code"] == JSONRPC_INVALID_REQUEST
|
||||
|
||||
|
||||
def test_wrong_jsonrpc_version_keeps_id():
|
||||
responses, _ = run('{"jsonrpc": "1.0", "id": 7, "method": "go"}\n')
|
||||
assert responses[0]["error"]["code"] == JSONRPC_INVALID_REQUEST
|
||||
assert responses[0]["id"] == 7
|
||||
|
||||
|
||||
def test_missing_method_string():
|
||||
responses, _ = run('{"jsonrpc": "2.0", "id": 1}\n')
|
||||
assert responses[0]["error"]["code"] == JSONRPC_INVALID_REQUEST
|
||||
|
||||
|
||||
def test_params_must_be_object_or_array():
|
||||
responses, _ = run('{"jsonrpc": "2.0", "id": 1, "method": "go", "params": 5}\n')
|
||||
assert responses[0]["error"]["code"] == JSONRPC_INVALID_PARAMS
|
||||
|
||||
|
||||
def test_unknown_method():
|
||||
responses, _ = run('{"jsonrpc": "2.0", "id": 1, "method": "nope"}\n')
|
||||
assert responses[0]["error"]["code"] == JSONRPC_METHOD_NOT_FOUND
|
||||
|
||||
|
||||
def test_successful_result_round_trip():
|
||||
responses, _ = run(
|
||||
'{"jsonrpc": "2.0", "id": 42, "method": "echo", "params": {"x": 1}}\n',
|
||||
{"echo": lambda params: params},
|
||||
)
|
||||
assert responses == [{"jsonrpc": "2.0", "id": 42, "result": {"x": 1}}]
|
||||
|
||||
|
||||
def test_rpc_error_is_forwarded_with_code_and_data():
|
||||
def handler(_params):
|
||||
raise RpcError(JSONRPC_INTERNAL_ERROR, "boom", data={"detail": "x"})
|
||||
|
||||
responses, _ = run(
|
||||
'{"jsonrpc": "2.0", "id": 1, "method": "go"}\n',
|
||||
{"go": handler},
|
||||
)
|
||||
assert responses[0]["error"] == {
|
||||
"code": JSONRPC_INTERNAL_ERROR,
|
||||
"message": "boom",
|
||||
"data": {"detail": "x"},
|
||||
}
|
||||
|
||||
|
||||
def test_unexpected_exception_becomes_internal_error():
|
||||
def handler(_params):
|
||||
raise ValueError("kaboom")
|
||||
|
||||
responses, stderr = run(
|
||||
'{"jsonrpc": "2.0", "id": 1, "method": "go"}\n',
|
||||
{"go": handler},
|
||||
)
|
||||
assert responses[0]["error"]["code"] == JSONRPC_INTERNAL_ERROR
|
||||
assert responses[0]["error"]["message"] == "kaboom"
|
||||
assert "Traceback" in stderr
|
||||
|
||||
|
||||
def test_notification_runs_handler_but_writes_no_response():
|
||||
calls: list[int] = []
|
||||
responses, _ = run(
|
||||
'{"jsonrpc": "2.0", "method": "go"}\n',
|
||||
{"go": lambda _params: calls.append(1)},
|
||||
)
|
||||
assert calls == [1]
|
||||
assert responses == []
|
||||
|
||||
|
||||
def test_blank_lines_skipped_and_requests_processed_in_order():
|
||||
line = (
|
||||
'{"jsonrpc": "2.0", "id": 1, "method": "go"}\n'
|
||||
"\n"
|
||||
" \n"
|
||||
'{"jsonrpc": "2.0", "id": 2, "method": "go"}\n'
|
||||
)
|
||||
responses, _ = run(line, {"go": lambda _params: "ok"})
|
||||
assert [r["id"] for r in responses] == [1, 2]
|
||||
@@ -0,0 +1,85 @@
|
||||
"""Tests for settings persistence (``bonsaiviewer_autodesk.settings``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from bonsaiviewer_autodesk import settings
|
||||
|
||||
|
||||
def _write_settings_json(config_dir, data: dict) -> None:
|
||||
(config_dir / "settings.json").write_text(json.dumps(data), encoding="utf-8")
|
||||
|
||||
|
||||
def test_config_root_uses_xdg_on_linux(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(settings.platform, "system", lambda: "Linux")
|
||||
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg"))
|
||||
root = settings.config_root()
|
||||
assert root == tmp_path / "xdg" / "bonsaiviewer-autodesk"
|
||||
assert root.is_dir()
|
||||
|
||||
|
||||
def test_config_root_uses_appdata_on_windows(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(settings.platform, "system", lambda: "Windows")
|
||||
monkeypatch.setenv("APPDATA", str(tmp_path / "appdata"))
|
||||
root = settings.config_root()
|
||||
assert root == tmp_path / "appdata" / "bonsaiviewer-autodesk"
|
||||
assert root.is_dir()
|
||||
|
||||
|
||||
def test_config_root_uses_application_support_on_macos(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(settings.platform, "system", lambda: "Darwin")
|
||||
monkeypatch.setattr(settings.Path, "home", lambda: tmp_path / "home")
|
||||
root = settings.config_root()
|
||||
assert root == tmp_path / "home" / "Library" / "Application Support" / "bonsaiviewer-autodesk"
|
||||
assert root.is_dir()
|
||||
|
||||
|
||||
def test_save_and_load_client_id_strips_whitespace(config_dir):
|
||||
settings.save_client_id(" abc123 ")
|
||||
assert settings.load_client_id() == "abc123"
|
||||
|
||||
|
||||
def test_load_client_id_empty_when_nothing_configured(config_dir):
|
||||
assert settings.load_client_id() == ""
|
||||
|
||||
|
||||
def test_callback_port_round_trips(config_dir):
|
||||
settings.save_callback_port(9001)
|
||||
assert settings.stored_callback_port() == 9001
|
||||
|
||||
|
||||
def test_callback_port_defaults_when_unset(config_dir):
|
||||
assert settings.stored_callback_port() == settings.DEFAULT_CALLBACK_PORT
|
||||
|
||||
|
||||
def test_callback_port_defaults_on_non_numeric_value(config_dir):
|
||||
_write_settings_json(config_dir, {"callback_port": "not-a-number"})
|
||||
assert settings.stored_callback_port() == settings.DEFAULT_CALLBACK_PORT
|
||||
|
||||
|
||||
def test_callback_port_defaults_on_out_of_range_value(config_dir):
|
||||
_write_settings_json(config_dir, {"callback_port": 70000})
|
||||
assert settings.stored_callback_port() == settings.DEFAULT_CALLBACK_PORT
|
||||
|
||||
|
||||
def test_save_callback_port_rejects_out_of_range(config_dir):
|
||||
with pytest.raises(ValueError):
|
||||
settings.save_callback_port(0)
|
||||
with pytest.raises(ValueError):
|
||||
settings.save_callback_port(70000)
|
||||
|
||||
|
||||
def test_corrupt_settings_file_is_treated_as_empty(config_dir):
|
||||
(config_dir / "settings.json").write_text("{ not valid json", encoding="utf-8")
|
||||
assert settings.load_client_id() == ""
|
||||
assert settings.stored_callback_port() == settings.DEFAULT_CALLBACK_PORT
|
||||
|
||||
|
||||
def test_save_client_id_preserves_other_keys(config_dir):
|
||||
settings.save_callback_port(9001)
|
||||
settings.save_client_id("abc")
|
||||
assert settings.stored_callback_port() == 9001
|
||||
assert settings.load_client_id() == "abc"
|
||||
@@ -30,8 +30,7 @@ Run
|
||||
bonsaiviewer-autodesk
|
||||
|
||||
The connector launches without configuration. On first run, open the settings
|
||||
dialog, or set the ``APS_CLIENT_ID`` environment variable, to configure the
|
||||
Autodesk client ID and OAuth callback port.
|
||||
dialog to configure the Autodesk client ID and OAuth callback port.
|
||||
|
||||
For direct protocol testing, send newline-delimited JSON-RPC 2.0 requests on
|
||||
standard input:
|
||||
@@ -51,11 +50,8 @@ exit.
|
||||
Configuration
|
||||
-------------
|
||||
|
||||
The connector reads the Autodesk client ID from these places, in order:
|
||||
|
||||
1. ``APS_CLIENT_ID`` environment variable.
|
||||
2. ``settings.json`` in the connector config directory, written by the settings
|
||||
dialog.
|
||||
The connector reads the Autodesk client ID from ``settings.json`` in the
|
||||
connector config directory, written by the settings dialog.
|
||||
|
||||
The OAuth callback host is always ``localhost``. The callback port defaults to
|
||||
``8080`` and can be changed in the settings dialog.
|
||||
|
||||
Reference in New Issue
Block a user