bambuddy/backend/tests/unit/services/test_model_provider_interface.py

"""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.fixture(autouse=True)
    def _auth_off(self):
        """These describe the auth-off install, the only one whose sign-in is
        stored without a user (see the auth-on test below)."""
        with patch(
            "backend.app.services.model_providers.makerworld.provider.is_auth_enabled",
            AsyncMock(return_value=False),
        ):
            yield

    @pytest.mark.asyncio
    async def test_with_auth_on_no_identity_borrows_no_stored_sign_in(self):
        """With auth on, a caller without a user (an API key without Allow
        Cloud Access) gets no token: the sign-in stored without a user is the
        auth-off install's, and may be left over after auth was turned on."""
        stored = AsyncMock(return_value=("global-tok", "admin@x.com", "global"))
        with (
            patch(
                "backend.app.services.model_providers.makerworld.provider.is_auth_enabled",
                AsyncMock(return_value=True),
            ),
            patch("backend.app.services.model_providers.makerworld.provider.get_stored_token", stored),
        ):
            svc = await makerworld_provider.build_service(db=AsyncMock(), user=None)

        assert svc._auth_token is None
        stored.assert_not_awaited()
        await svc.close()

    @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()