content-find-agent-new/tests/test_oss_upload.py

from __future__ import annotations

import httpx

from content_agent.integrations.oss_upload import upload_video_from_url


class FakeResponse:
    def __init__(self, body, *, status_code=200):
        self._body = body
        self.status_code = status_code
        self.request = httpx.Request("POST", "http://oss.test/upload")

    def raise_for_status(self):
        if self.status_code >= 400:
            response = httpx.Response(self.status_code, request=self.request)
            raise httpx.HTTPStatusError("bad", request=self.request, response=response)

    def json(self):
        return self._body


def test_upload_video_from_url_parses_oss_object():
    seen = {}

    def post(*args, **kwargs):
        seen.update({"args": args, **kwargs})
        return FakeResponse(
            {
                "status": 0,
                "msg": "ok",
                "oss_object": {
                    "cdn_url": "https://res.example/video.mp4",
                    "oss_object_key": "crawler/video/a.mp4",
                    "save_oss_timestamp": 123,
                },
            }
        )

    result = upload_video_from_url(
        "https://source.example/video.mp4",
        project="content-agent",
        referer={"Referer": "https://www.douyin.com/"},
        http_post=post,
        endpoint="http://oss.test/upload",
    )

    assert result["status"] == "ok"
    assert result["oss_url"] == "https://res.example/video.mp4"
    assert result["oss_object_key"] == "crawler/video/a.mp4"
    assert seen["json"]["src_type"] == "video"
    assert seen["json"]["use_proxy"] is True
    assert seen["json"]["project"] == "content-agent"
    assert "referer" not in seen["json"]
    assert result["oss_payload_mode"] == "no_referer"


def test_upload_video_from_url_failure_is_returned_not_raised():
    result = upload_video_from_url(
        "https://source.example/video.mp4",
        http_post=lambda *a, **k: FakeResponse({}, status_code=500),
        endpoint="http://oss.test/upload",
    )

    assert result["status"] == "failed"
    assert result["failure_type"] == "oss_upload_http_error"
    assert result["http_status_code"] == 500
    assert result["error_message"] == "bad"
    assert result["oss_payload_mode"] == "no_referer"


def test_upload_video_from_url_invalid_response_keeps_summary():
    result = upload_video_from_url(
        "https://source.example/video.mp4",
        http_post=lambda *a, **k: FakeResponse(
            {"status": 10000, "msg": "bad", "oss_object": None}
        ),
        endpoint="http://oss.test/upload",
    )

    assert result["status"] == "failed"
    assert result["failure_type"] == "oss_upload_response_invalid"
    assert result["oss_payload_mode"] == "no_referer"
    assert result["oss_response_summary"] == {
        "status": 10000,
        "msg": "bad",
        "oss_object_present": False,
        "oss_object_has_cdn_url": False,
    }