Create-knowledge-find-decode/tests/test_oss.py

"""OSS 转存离线测试：纯 parser + upload_stream + to_oss 兜底（注入 http_post）。"""
from __future__ import annotations

import httpx
import pytest

from core.config import PgConfig, Settings
from acquisition.oss import (
    OssError, parse_oss_response, to_oss, upload_stream,
)


def _settings() -> Settings:
    return Settings(
        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
        aiddit_crawler_base_url="x", crawler_timeout=30,
        openrouter_timeout_seconds=90,
        openrouter_model="m",
        openrouter_base_url="b", openrouter_api_key="k",
        llm_model="m", max_cards=12, frames_dir="f", douyin_ratio="540p", data_dir="",
        oss_upload_url="http://oss.test/upload_stream", oss_upload_timeout_seconds=60)


class _Resp:
    def __init__(self, payload): self._p = payload
    def raise_for_status(self): return None
    def json(self): return self._p


def test_parse_ok():
    assert parse_oss_response(
        {"status": 0, "oss_object": {"cdn_url": "https://res/x.mp4"}}) == "https://res/x.mp4"


def test_parse_status_error():
    with pytest.raises(OssError):
        parse_oss_response({"status": 1, "msg": "fail"})


def test_parse_missing_cdn():
    with pytest.raises(OssError):
        parse_oss_response({"status": 0, "oss_object": {}})


def test_upload_stream_posts_and_returns_cdn():
    captured = {}

    def post(url, json=None, headers=None, timeout=None):
        captured["url"] = url
        captured["json"] = json
        return _Resp({"status": 0, "oss_object": {"cdn_url": "https://res/y.webp"}})

    cdn = upload_stream("http://src/y.webp", src_type="image",
                        settings=_settings(), http_post=post)
    assert cdn == "https://res/y.webp"
    assert captured["url"] == "http://oss.test/upload_stream"
    assert captured["json"] == {"src_url": "http://src/y.webp", "src_type": "image"}


def test_upload_stream_bad_src_type():
    with pytest.raises(OssError):
        upload_stream("http://src/a", src_type="audio", settings=_settings())


def test_to_oss_fallback_on_http_error():
    def post(url, json=None, headers=None, timeout=None):
        raise httpx.ConnectError("boom")

    # 失败返回原 url，永不抛
    assert to_oss("http://src/z.mp4", "video",
                  settings=_settings(), http_post=post) == "http://src/z.mp4"


def test_to_oss_fallback_on_business_error():
    def post(url, json=None, headers=None, timeout=None):
        return _Resp({"status": 500, "msg": "boom"})

    assert to_oss("http://src/z.mp4", "video",
                  settings=_settings(), http_post=post) == "http://src/z.mp4"


def test_to_oss_success():
    def post(url, json=None, headers=None, timeout=None):
        return _Resp({"status": 0, "oss_object": {"cdn_url": "https://res/z.mp4"}})

    assert to_oss("http://src/z.mp4", "video",
                  settings=_settings(), http_post=post) == "https://res/z.mp4"


def test_to_oss_empty_url_passthrough():
    assert to_oss("", "image", settings=_settings()) == ""