bambuddy/backend/tests/integration/test_cloud_token_auth_migration.py

"""Cloud-credential migration across the auth on/off boundary (#2530).

``get_stored_token`` reads global ``Settings`` rows when auth is disabled and
``User.cloud_token`` when it's enabled. Toggling auth therefore switches which
store the ``/cloud/*`` routes consult. Without an explicit hand-off the token
is stranded in the store nobody reads: ``build_authenticated_cloud`` returns
``None``, Phase 2 of ``get_filament_info`` is skipped entirely, and the caller
sees a ``200`` full of local-preset fallbacks with no sign the cloud was never
contacted. That silent degradation is what #2530 actually reported.

These tests pin the hand-off in both directions, and — just as importantly —
pin the two cases where Bambuddy must refuse to guess who owns a credential.
"""

import logging

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.api.routes import cloud as cloud_routes
from backend.app.core.auth import get_password_hash
from backend.app.models.settings import Settings
from backend.app.models.user import User
from backend.app.services.bambu_cloud import BambuCloudError
from backend.app.services.bambu_cloud_credentials import (
    CLOUD_EMAIL_KEY,
    CLOUD_REGION_KEY,
    CLOUD_TOKEN_KEY,
    get_stored_token,
)


async def _seed_global_token(db: AsyncSession, token: str = "tok-global", region: str = "china") -> None:
    db.add(Settings(key=CLOUD_TOKEN_KEY, value=token))
    db.add(Settings(key=CLOUD_EMAIL_KEY, value="owner@example.com"))
    db.add(Settings(key=CLOUD_REGION_KEY, value=region))
    await db.commit()


async def _global_rows(db: AsyncSession) -> dict[str, str]:
    rows = (
        (
            await db.execute(
                select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY, CLOUD_REGION_KEY]))
            )
        )
        .scalars()
        .all()
    )
    return {r.key: r.value for r in rows}


async def _make_admin(db: AsyncSession, username: str) -> User:
    user = User(
        username=username,
        password_hash=get_password_hash("AdminPass1!"),
        role="admin",
        is_active=True,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user


# ---------------------------------------------------------------------------
# auth OFF -> ON
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_setup_migrates_global_token_to_created_admin(async_client: AsyncClient, db_session: AsyncSession):
    """The reporter's exact path: link cloud with auth off, then enable auth."""
    await _seed_global_token(db_session)

    resp = await async_client.post(
        "/api/v1/auth/setup",
        json={"auth_enabled": True, "admin_username": "admin", "admin_password": "AdminPass1!"},
    )
    assert resp.status_code == 200, resp.text

    admin = (await db_session.execute(select(User).where(User.role == "admin"))).scalar_one()
    token, email, region = await get_stored_token(db_session, admin)
    assert token == "tok-global"
    assert email == "owner@example.com"
    assert region == "china", "region must survive the hop, not silently reset to global"

    # Credential must not be left at rest in a table nothing reads any more.
    assert await _global_rows(db_session) == {}


@pytest.mark.asyncio
async def test_setup_migrates_to_sole_pre_existing_admin(async_client: AsyncClient, db_session: AsyncSession):
    """Re-enabling auth when exactly one admin already exists has one obvious owner."""
    admin = await _make_admin(db_session, "solo")
    await _seed_global_token(db_session, token="tok-solo")

    resp = await async_client.post("/api/v1/auth/setup", json={"auth_enabled": True})
    assert resp.status_code == 200, resp.text
    assert resp.json()["admin_created"] is False

    await db_session.refresh(admin)
    token, _, _ = await get_stored_token(db_session, admin)
    assert token == "tok-solo"
    assert await _global_rows(db_session) == {}


@pytest.mark.asyncio
async def test_setup_refuses_to_guess_owner_when_multiple_admins(async_client: AsyncClient, db_session: AsyncSession):
    """Two admins, one credential: handing it to either is a security decision we don't make."""
    a = await _make_admin(db_session, "admin_a")
    b = await _make_admin(db_session, "admin_b")
    await _seed_global_token(db_session, token="tok-ambiguous")

    resp = await async_client.post("/api/v1/auth/setup", json={"auth_enabled": True})
    assert resp.status_code == 200, resp.text

    await db_session.refresh(a)
    await db_session.refresh(b)
    assert a.cloud_token is None
    assert b.cloud_token is None
    # Left intact so the operator can re-link rather than lose it.
    assert (await _global_rows(db_session))[CLOUD_TOKEN_KEY] == "tok-ambiguous"


@pytest.mark.asyncio
async def test_setup_with_auth_disabled_leaves_global_token_untouched(
    async_client: AsyncClient, db_session: AsyncSession
):
    """Completing setup while declining auth must not move anything."""
    await _seed_global_token(db_session, token="tok-stay")

    resp = await async_client.post("/api/v1/auth/setup", json={"auth_enabled": False})
    assert resp.status_code == 200, resp.text

    assert (await _global_rows(db_session))[CLOUD_TOKEN_KEY] == "tok-stay"


# ---------------------------------------------------------------------------
# auth ON -> OFF
# ---------------------------------------------------------------------------


async def _admin_bearer(async_client: AsyncClient, username: str = "admin") -> str:
    await async_client.post(
        "/api/v1/auth/setup",
        json={"auth_enabled": True, "admin_username": username, "admin_password": "AdminPass1!"},
    )
    login = await async_client.post(
        "/api/v1/auth/login",
        json={"username": username, "password": "AdminPass1!"},
    )
    return login.json()["access_token"]


@pytest.mark.asyncio
async def test_disable_auth_migrates_admin_token_to_global(async_client: AsyncClient, db_session: AsyncSession):
    bearer = await _admin_bearer(async_client)
    admin = (await db_session.execute(select(User).where(User.role == "admin"))).scalar_one()
    admin.cloud_token = "tok-user"
    admin.cloud_email = "user@example.com"
    admin.cloud_region = "china"
    await db_session.commit()

    resp = await async_client.post("/api/v1/auth/disable", headers={"Authorization": f"Bearer {bearer}"})
    assert resp.status_code == 200, resp.text

    rows = await _global_rows(db_session)
    assert rows[CLOUD_TOKEN_KEY] == "tok-user"
    assert rows[CLOUD_REGION_KEY] == "china"

    await db_session.refresh(admin)
    assert admin.cloud_token is None, "credential must not be duplicated across both stores"

    # And the no-auth read path now finds it.
    token, _, _ = await get_stored_token(db_session, None)
    assert token == "tok-user"


@pytest.mark.asyncio
async def test_disable_auth_does_not_clobber_existing_global_token(async_client: AsyncClient, db_session: AsyncSession):
    """A stale global row is still somebody's credential — refuse rather than overwrite."""
    bearer = await _admin_bearer(async_client)
    admin = (await db_session.execute(select(User).where(User.role == "admin"))).scalar_one()
    admin.cloud_token = "tok-user"
    await db_session.commit()
    await _seed_global_token(db_session, token="tok-preexisting")

    resp = await async_client.post("/api/v1/auth/disable", headers={"Authorization": f"Bearer {bearer}"})
    assert resp.status_code == 200, resp.text

    assert (await _global_rows(db_session))[CLOUD_TOKEN_KEY] == "tok-preexisting"
    await db_session.refresh(admin)
    assert admin.cloud_token == "tok-user", "admin keeps their token when we decline to migrate"


@pytest.mark.asyncio
async def test_disable_auth_with_no_cloud_token_is_a_noop(async_client: AsyncClient, db_session: AsyncSession):
    bearer = await _admin_bearer(async_client)

    resp = await async_client.post("/api/v1/auth/disable", headers={"Authorization": f"Bearer {bearer}"})
    assert resp.status_code == 200, resp.text
    assert await _global_rows(db_session) == {}


# ---------------------------------------------------------------------------
# Log-level classification for cloud preset misses (#2530)
# ---------------------------------------------------------------------------


def test_bambu_cloud_error_carries_status_code():
    """The 400-vs-fault distinction depends on this attribute existing."""
    from backend.app.services.bambu_cloud import BambuCloudError

    assert BambuCloudError("boom").status_code is None
    assert BambuCloudError("boom", status_code=400).status_code == 400
    # Every pre-existing raise site passes a bare message; must not break.
    assert str(BambuCloudError("Request failed: timeout")) == "Request failed: timeout"


class _StubCloud:
    """Minimal stand-in for BambuCloudService that fails every preset lookup."""

    def __init__(self, error: Exception):
        self._error = error
        self.is_authenticated = True

    async def get_setting_detail(self, setting_id: str) -> dict:
        raise self._error

    async def close(self) -> None:
        return None


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("error", "expected_level"),
    [
        (BambuCloudError("missing", status_code=400), logging.DEBUG),
        (BambuCloudError("unauthorized", status_code=401), logging.WARNING),
        (BambuCloudError("bad gateway", status_code=502), logging.WARNING),
        (BambuCloudError("Request failed: timeout"), logging.WARNING),
    ],
    ids=["expected-400-miss", "expired-token", "cloud-outage", "transport-failure"],
)
async def test_preset_miss_logs_at_debug_but_faults_stay_at_warning(
    monkeypatch, caplog, db_session: AsyncSession, error: Exception, expected_level: int
):
    """A preset the catalog doesn't carry is routine; an expired token is a fault.

    Drives the real ``get_filament_info`` route so the classification is exercised
    where it lives, not re-derived in the test.
    """
    monkeypatch.setattr(cloud_routes, "build_authenticated_cloud", _stub_builder(error))
    # Phase 1 would otherwise short-circuit the cloud call on a warm cache.
    monkeypatch.setattr(cloud_routes, "_filament_cache", {})

    with caplog.at_level(logging.DEBUG, logger="backend.app.api.routes.cloud"):
        result = await cloud_routes.get_filament_info(setting_ids=["GFL05"], db=db_session, current_user=None)

    records = [r for r in caplog.records if "Failed to get cloud preset" in r.getMessage()]
    assert len(records) == 1, "the miss must be logged exactly once"
    assert records[0].levelno == expected_level
    assert "GFL05" in records[0].getMessage()
    assert "GFSL05" in records[0].getMessage(), "the translated API ID stays in the message"

    # Whatever the level, the endpoint still answers and falls through to Phase 3.
    assert isinstance(result, dict)


def _stub_builder(error: Exception):
    async def _build(db, user):
        return _StubCloud(error)

    return _build