SmartUp/backend/test_custom_pages_refresh_auth.py

import asyncio
import json
import sys
from pathlib import Path

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

sys.path.insert(0, str(Path(__file__).resolve().parent))

from app.database import Base
from app.models.custom_page import CustomPage
from app.models.upstream import Upstream
from app.routers import custom_pages


@pytest.fixture()
def db_session():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)


def _linked_page(db, upstream: Upstream) -> CustomPage:
    db.add(upstream)
    db.commit()
    db.refresh(upstream)
    page = CustomPage(
        name="Login",
        url="https://meow.example/login",
        access_mode="remote_browser",
        linked_upstream_id=upstream.id,
    )
    db.add(page)
    db.commit()
    db.refresh(page)
    return page


def _install_refresh_fakes(monkeypatch, candidates: list[dict], calls: list[dict]):
    monkeypatch.setattr(custom_pages.browser_sessions, "find_by_page_id", lambda _pid: object())

    async def fake_extract_all(_session):
        return {"candidates": candidates}

    monkeypatch.setattr(custom_pages, "extract_all", fake_extract_all)

    class FakeUpstreamClient:
        def __init__(self, **kwargs):
            self.kwargs = kwargs

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return False

        def get_available_groups(self, endpoint):
            calls.append({"endpoint": endpoint, **self.kwargs})
            return [{"id": "default", "name": "Default"}]

    import app.services.upstream_client as upstream_client_module

    monkeypatch.setattr(upstream_client_module, "UpstreamClient", FakeUpstreamClient)


def test_refresh_auth_sub2api_prefers_bearer_over_cookie_bundle(db_session, monkeypatch):
    upstream = Upstream(
        name="Meow",
        base_url="https://api.saki.lat",
        api_prefix="/api/v1",
        auth_type="login_password",
        auth_config_json=json.dumps({
            "email": "alice@example.test",
            "password": "secret",
            "login_path": "/auth/login",
        }),
        groups_endpoint="/groups/available",
        rate_endpoint="/groups/rates",
    )
    page = _linked_page(db_session, upstream)
    calls: list[dict] = []
    _install_refresh_fakes(monkeypatch, [
        {"type": "cookie_bundle", "value": "cf_clearance=cf; session=s", "cookie_count": 2},
        {"type": "bearer_token", "value": "jwt.header.payload", "source": "localStorage.auth_token"},
    ], calls)

    response = asyncio.run(custom_pages.refresh_auth(page.id, db_session, object()))

    db_session.refresh(upstream)
    cfg = json.loads(upstream.auth_config_json)
    assert response.success is True
    assert upstream.auth_type == "bearer"
    assert cfg["token"] == "jwt.header.payload"
    assert "cookie_string" not in cfg
    assert upstream.api_prefix == "/api/v1"
    assert upstream.groups_endpoint == "/groups/available"
    assert calls[0]["auth_type"] == "bearer"
    assert calls[0]["endpoint"] == "/groups/available"


def test_refresh_auth_sub2api_rejects_cookie_only_capture(db_session, monkeypatch):
    upstream = Upstream(
        name="Meow",
        base_url="https://api.saki.lat",
        api_prefix="/api/v1",
        auth_type="login_password",
        auth_config_json=json.dumps({"login_path": "/auth/login"}),
        groups_endpoint="/groups/available",
        rate_endpoint="/groups/rates",
    )
    page = _linked_page(db_session, upstream)
    _install_refresh_fakes(monkeypatch, [
        {"type": "cookie_bundle", "value": "cf_clearance=cf; session=s", "cookie_count": 2},
    ], [])

    response = asyncio.run(custom_pages.refresh_auth(page.id, db_session, object()))

    db_session.refresh(upstream)
    cfg = json.loads(upstream.auth_config_json)
    assert response.success is False
    assert "Sub2API 需要 Bearer Token" in response.message
    assert upstream.auth_type == "login_password"
    assert "cookie_string" not in cfg


def test_refresh_auth_new_api_user_uses_cookie_bundle_and_resets_user_endpoints(db_session, monkeypatch):
    upstream = Upstream(
        name="New API User",
        base_url="https://newapi.example",
        api_prefix="/api/v1",
        auth_type="login_password",
        auth_config_json=json.dumps({
            "email": "alice",
            "password": "secret",
            "login_path": "/api/user/login",
        }),
        groups_endpoint="/groups/available",
        rate_endpoint="/groups/rates",
    )
    page = _linked_page(db_session, upstream)
    calls: list[dict] = []
    _install_refresh_fakes(monkeypatch, [
        {
            "type": "cookie_bundle",
            "value": "cf_clearance=cf; session=s",
            "cookie_count": 2,
            "new_api_user": "42",
        },
        {"type": "bearer_token", "value": "jwt.header.payload"},
    ], calls)

    response = asyncio.run(custom_pages.refresh_auth(page.id, db_session, object()))

    db_session.refresh(upstream)
    cfg = json.loads(upstream.auth_config_json)
    assert response.success is True
    assert upstream.auth_type == "cookie"
    assert cfg["cookie_string"] == "cf_clearance=cf; session=s"
    assert cfg["new_api_user"] == "42"
    assert upstream.api_prefix == ""
    assert upstream.groups_endpoint == "/api/user/self/groups"
    assert upstream.rate_endpoint == "/api/user/self/groups"
    assert calls[0]["auth_type"] == "cookie"
    assert calls[0]["endpoint"] == "/api/user/self/groups"