mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
222 lines
10 KiB
Python
222 lines
10 KiB
Python
"""Tests for the MakerWorld provider descriptor (``provider.py``).
|
|
|
|
The descriptor is the interface between the route layer and the per-request
|
|
service: identity, URL routing, auth requirements, and the factory that seeds
|
|
a service with the caller's stored Bambu Cloud bearer token.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.core.permissions import Permission
|
|
from backend.app.services.model_providers import makerworld_provider
|
|
from backend.app.services.model_providers.base import ModelProvider
|
|
from backend.app.services.model_providers.makerworld.service import MakerWorldService
|
|
|
|
|
|
class TestMakerWorldProviderDescriptor:
|
|
def test_identity_fields(self):
|
|
assert makerworld_provider.source_type == "makerworld"
|
|
assert makerworld_provider.display_name == "MakerWorld"
|
|
assert makerworld_provider.host_patterns == ("makerworld.com",)
|
|
assert makerworld_provider.default_folder_name == "MakerWorld"
|
|
|
|
def test_permissions(self):
|
|
assert makerworld_provider.view_permission == Permission.MAKERWORLD_VIEW
|
|
assert makerworld_provider.import_permission == Permission.MAKERWORLD_IMPORT
|
|
|
|
def test_auth_descriptor(self):
|
|
assert makerworld_provider.auth is not None
|
|
assert makerworld_provider.auth.auth_type == "bambu_cloud_bearer"
|
|
assert makerworld_provider.auth.credential_fields == ()
|
|
|
|
def test_supports_url_host_suffix_match(self):
|
|
assert makerworld_provider.supports_url("https://makerworld.com/en/models/1400373")
|
|
assert makerworld_provider.supports_url("https://www.makerworld.com/models/1400373")
|
|
assert makerworld_provider.supports_url("makerworld.com/models/1")
|
|
|
|
def test_rejects_foreign_hosts_and_garbage(self):
|
|
assert not makerworld_provider.supports_url("https://thingiverse.com/thing/123")
|
|
assert not makerworld_provider.supports_url("https://makerworld.com.evil.example/x")
|
|
assert not makerworld_provider.supports_url("")
|
|
assert not makerworld_provider.supports_url(None) # type: ignore[arg-type]
|
|
assert not makerworld_provider.supports_url(123) # type: ignore[arg-type]
|
|
|
|
def test_thumbnail_hosts_is_the_cdn_allowlist(self):
|
|
assert "makerworld.bblmw.com" in makerworld_provider.thumbnail_hosts()
|
|
assert "public-cdn.bblmw.com" in makerworld_provider.thumbnail_hosts()
|
|
|
|
def test_download_hosts_is_the_cdn_allowlist(self):
|
|
"""The download-guard SSRF seam mirrors the thumbnail one — a provider
|
|
that fetches files server-side declares the hosts its service may
|
|
fetch from (review round 3, note 2)."""
|
|
assert "makerworld.bblmw.com" in makerworld_provider.download_hosts()
|
|
assert "public-cdn.bblmw.com" in makerworld_provider.download_hosts()
|
|
|
|
def test_parse_and_canonical_roundtrip(self):
|
|
ref = makerworld_provider.parse_url("https://makerworld.com/en/models/1400373-slug#profileId-1452154")
|
|
assert ref.external_id == "1400373"
|
|
assert ref.sub_id == "1452154"
|
|
assert ref.source_type == "makerworld"
|
|
assert makerworld_provider.canonical_url(ref) == "https://makerworld.com/models/1400373#profileId-1452154"
|
|
|
|
def test_canonical_plate_less_shape(self):
|
|
ref = makerworld_provider.parse_url("https://makerworld.com/models/999")
|
|
assert makerworld_provider.canonical_url(ref) == "https://makerworld.com/models/999"
|
|
|
|
def test_source_url_filter_matches_model_and_any_plate(self):
|
|
"""The already-imported predicate must cover the whole-model key and
|
|
every per-plate key — MakerWorld's canonical shape appends
|
|
``#profileId-{n}`` (review round 3: shape knowledge belongs to the
|
|
provider, not the route)."""
|
|
from sqlalchemy import String, column as sa_column
|
|
|
|
expr = makerworld_provider.source_url_filter(sa_column("source_url", String), "1400373")
|
|
sql = str(expr.compile(compile_kwargs={"literal_binds": True}))
|
|
assert "source_url = 'https://makerworld.com/models/1400373'" in sql
|
|
assert "LIKE 'https://makerworld.com/models/1400373#profileId-%'" in sql
|
|
|
|
def test_default_source_url_filter_is_exact_match_only(self):
|
|
"""A provider without plate-shaped keys inherits the default: exact
|
|
match on its whole-model canonical URL, nothing else."""
|
|
from sqlalchemy import String, column as sa_column
|
|
|
|
expr = _WholeModelProvider().source_url_filter(sa_column("source_url", String), "42")
|
|
sql = str(expr.compile(compile_kwargs={"literal_binds": True}))
|
|
assert sql == "source_url = 'https://example.com/models/42'"
|
|
assert "LIKE" not in sql
|
|
|
|
|
|
class _WholeModelProvider(ModelProvider):
|
|
"""Minimal concrete provider that keys dedupe at whole-model granularity
|
|
only — exercises the inherited default ``source_url_filter``."""
|
|
|
|
source_type = "wholemodel"
|
|
display_name = "WholeModel"
|
|
|
|
async def build_service(self, *, db, user, api_key_owner=None, client=None):
|
|
raise NotImplementedError
|
|
|
|
def parse_url(self, url):
|
|
raise NotImplementedError
|
|
|
|
def canonical_url(self, ref):
|
|
return f"https://example.com/models/{ref.external_id}"
|
|
|
|
|
|
class TestBuildService:
|
|
"""build_service must reproduce exactly what the old route helper did:
|
|
read the caller's stored Bambu Cloud token and wire the rejected-token
|
|
callback so a 401 invalidates the shared credential app-wide."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seeds_token_and_auth_failure_callback(self):
|
|
db = AsyncMock()
|
|
user = AsyncMock()
|
|
user.id = 7
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.services.model_providers.makerworld.provider.get_stored_token",
|
|
AsyncMock(return_value=("tok-abc", "e@x.com", "global")),
|
|
),
|
|
patch(
|
|
"backend.app.services.model_providers.makerworld.provider.mark_cloud_token_invalid",
|
|
AsyncMock(),
|
|
) as mark_invalid,
|
|
):
|
|
svc = await makerworld_provider.build_service(db=db, user=user)
|
|
assert isinstance(svc, MakerWorldService)
|
|
assert svc._auth_token == "tok-abc"
|
|
assert svc._user is user
|
|
await svc._on_auth_failure()
|
|
mark_invalid.assert_awaited_once_with(7)
|
|
await svc.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anonymous_user_still_gets_auth_callback(self):
|
|
"""No user ≠ nothing to invalidate. Auth-disabled single-user installs
|
|
hold their token in global Settings (``get_stored_token(db, None)``
|
|
reads it), so a rejection must still be recorded — ``user_id=None``
|
|
writes the global flag (review blocker 5). The callback stays wired;
|
|
only a *stray* non-expiry 401 keeps it a no-op."""
|
|
db = AsyncMock()
|
|
with (
|
|
patch(
|
|
"backend.app.services.model_providers.makerworld.provider.get_stored_token",
|
|
AsyncMock(return_value=(None, None, "global")),
|
|
),
|
|
patch(
|
|
"backend.app.services.model_providers.makerworld.provider.mark_cloud_token_invalid",
|
|
AsyncMock(),
|
|
) as mark_invalid,
|
|
):
|
|
svc = await makerworld_provider.build_service(db=db, user=None)
|
|
|
|
assert svc._auth_token is None
|
|
assert svc._on_auth_failure is not None
|
|
# Firing it records the *global* flag (user_id=None), not a per-user row.
|
|
await svc._on_auth_failure()
|
|
mark_invalid.assert_awaited_once_with(None)
|
|
await svc.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_service_passes_declared_thumbnail_hosts(self):
|
|
"""The SSRF seam contract: ``fetch_thumbnail``'s allowlist must come
|
|
from ``ModelProvider.thumbnail_hosts()`` via build_service — not from
|
|
a hardcoded copy inside the service (review round 2, item 1)."""
|
|
db = AsyncMock()
|
|
with patch(
|
|
"backend.app.services.model_providers.makerworld.provider.get_stored_token",
|
|
AsyncMock(return_value=(None, None, "global")),
|
|
):
|
|
svc = await makerworld_provider.build_service(db=db, user=None)
|
|
|
|
assert svc._thumbnail_hosts == makerworld_provider.thumbnail_hosts()
|
|
assert len(svc._thumbnail_hosts) > 0
|
|
await svc.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_service_passes_declared_download_hosts(self):
|
|
"""Symmetric SSRF seam contract: ``download``'s allowlist must come
|
|
from ``ModelProvider.download_hosts()`` via build_service — not from
|
|
a hardcoded copy inside the service (review round 3, note 2)."""
|
|
db = AsyncMock()
|
|
with patch(
|
|
"backend.app.services.model_providers.makerworld.provider.get_stored_token",
|
|
AsyncMock(return_value=(None, None, "global")),
|
|
):
|
|
svc = await makerworld_provider.build_service(db=db, user=None)
|
|
|
|
assert svc._download_hosts == makerworld_provider.download_hosts()
|
|
assert len(svc._download_hosts) > 0
|
|
await svc.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_key_owner_is_the_fallback_identity(self):
|
|
"""API-keyed callers carry identity on the key (#1777) — build_service
|
|
must use the key's owner when ``user`` is None."""
|
|
db = AsyncMock()
|
|
owner = AsyncMock()
|
|
owner.id = 11
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.services.model_providers.makerworld.provider.get_stored_token",
|
|
AsyncMock(return_value=("owner-tok", "owner@x.com", "global")),
|
|
),
|
|
patch(
|
|
"backend.app.services.model_providers.makerworld.provider.mark_cloud_token_invalid",
|
|
AsyncMock(),
|
|
) as mark_invalid,
|
|
):
|
|
svc = await makerworld_provider.build_service(db=db, user=None, api_key_owner=owner)
|
|
assert svc._auth_token == "owner-tok"
|
|
assert svc._user is owner
|
|
await svc._on_auth_failure()
|
|
mark_invalid.assert_awaited_once_with(11)
|
|
await svc.close()
|