353 lines
14 KiB
Python
353 lines
14 KiB
Python
"""COS upload contract tests: real SDK configuration, no external requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import time
|
|
import traceback
|
|
from collections.abc import Iterator
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
from qcloud_cos import CosClientError, CosConfig, CosS3Client
|
|
|
|
from doctor_workstation.core.errors import ApiBusinessError, ApiProtocolError, ApiTransportError
|
|
from doctor_workstation.services.api_client import ApiClient
|
|
from doctor_workstation.services.cos_upload import upload_video_to_cos
|
|
|
|
BUCKET = "synthetic-video-1234567890"
|
|
REGION = "ap-guangzhou"
|
|
PREFIX = "uploads/video/20261105/"
|
|
_AUTOMATIC = object()
|
|
|
|
|
|
def _credentials() -> dict[str, Any]:
|
|
now = int(time.time())
|
|
return {
|
|
"provider": "qcloud",
|
|
"fallback": False,
|
|
"bucket": BUCKET,
|
|
"region": REGION,
|
|
"key_prefix": PREFIX,
|
|
"object_key": "",
|
|
"host": "https://untrusted-host.example.test",
|
|
"cdn_domain": "https://cdn.example.test",
|
|
"max_size": 2 * 1024 * 1024 * 1024,
|
|
"start_time": now - 10,
|
|
"expired_time": now + 1800,
|
|
"credentials": {
|
|
"tmpSecretId": "synthetic-temporary-id",
|
|
"tmpSecretKey": "synthetic-temporary-secret",
|
|
"sessionToken": "synthetic-temporary-token",
|
|
},
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class _Harness:
|
|
source: Path
|
|
credentials: Any = field(default_factory=_credentials)
|
|
confirmation: Any = _AUTOMATIC
|
|
sdk_error: Exception | None = None
|
|
sdk_reports_progress: bool = True
|
|
api_failure: str = ""
|
|
log_sensitive_sdk_data: bool = False
|
|
api_requests: list[httpx.Request] = field(default_factory=list)
|
|
sdk_calls: list[dict[str, Any]] = field(default_factory=list)
|
|
configs: list[CosConfig] = field(default_factory=list)
|
|
progress: list[tuple[int, int]] = field(default_factory=list)
|
|
events: list[str] = field(default_factory=list)
|
|
|
|
def handle(self, request: httpx.Request) -> httpx.Response:
|
|
self.api_requests.append(request)
|
|
assert request.method == "POST"
|
|
assert request.headers["token"] == "synthetic-admin-token"
|
|
endpoint = request.url.path.rsplit("/", 1)[-1]
|
|
assert endpoint in {"ossCredentials", "ossConfirm"}
|
|
self.events.append(endpoint)
|
|
if self.api_failure == endpoint:
|
|
return httpx.Response(200, json={"code": 0, "msg": "synthetic API rejection"})
|
|
if endpoint == "ossCredentials":
|
|
return httpx.Response(200, json={"code": 1, "data": self.credentials})
|
|
assert request.extensions["timeout"]["read"] == 120.0
|
|
body = json.loads(request.content)
|
|
result = self.confirmation
|
|
if result is _AUTOMATIC:
|
|
address = "https://cdn.example.test/" + body["key"]
|
|
result = {"id": 19, "uri": address, "url": address, "type": 20, "cid": 0}
|
|
return httpx.Response(200, json={"code": 1, "data": result})
|
|
|
|
def client(self) -> ApiClient:
|
|
return ApiClient(
|
|
"https://api.example.test",
|
|
token="synthetic-admin-token",
|
|
max_retries=5,
|
|
transport=httpx.MockTransport(self.handle),
|
|
)
|
|
|
|
def processing(self) -> None:
|
|
self.events.append("processing")
|
|
|
|
def report(self, sent: int, total: int) -> None:
|
|
self.progress.append((sent, total))
|
|
|
|
|
|
@pytest.fixture
|
|
def harness(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[_Harness]:
|
|
source = tmp_path / "synthetic-replay.mp4"
|
|
source.write_bytes(b"synthetic-video-bytes")
|
|
state = _Harness(source)
|
|
|
|
def sdk_upload(sdk: CosS3Client, **kwargs: Any) -> dict[str, str]:
|
|
state.configs.append(sdk.get_conf())
|
|
state.sdk_calls.append(kwargs)
|
|
state.events.append("sdk-start")
|
|
if state.log_sensitive_sdk_data:
|
|
logging.getLogger("qcloud_cos.cos_auth").debug("synthetic-temporary-secret")
|
|
logging.getLogger("qcloud_cos.cos_client").error("synthetic-temporary-token")
|
|
if state.sdk_error is not None:
|
|
raise state.sdk_error
|
|
if state.sdk_reports_progress:
|
|
size = source.stat().st_size
|
|
callback = kwargs["progress_callback"]
|
|
callback(size // 2, size)
|
|
callback(size, size)
|
|
callback(size * 2, size) # Defensive clamping must keep binding pending.
|
|
state.events.append("sdk-done")
|
|
return {"ETag": '"synthetic-etag"'}
|
|
|
|
def no_external_request(*_args: Any, **_kwargs: Any) -> None:
|
|
pytest.fail("SDK tests must never make an external request")
|
|
|
|
monkeypatch.setattr(CosS3Client, "upload_file", sdk_upload)
|
|
monkeypatch.setattr("requests.Session.request", no_external_request)
|
|
# Restore the logger after the test while exercising the helper's choice.
|
|
monkeypatch.setattr(logging.getLogger("qcloud_cos"), "propagate", True)
|
|
yield state
|
|
|
|
|
|
@pytest.mark.parametrize("size", [1, 1024, 5 * 1024 * 1024])
|
|
def test_official_sdk_config_and_confirm_order_keep_completion_pending(
|
|
harness: _Harness, size: int
|
|
) -> None:
|
|
harness.source.write_bytes(b"v" * size)
|
|
harness.sdk_reports_progress = size > 1024 * 1024
|
|
with harness.client() as client:
|
|
result = upload_video_to_cos(
|
|
client,
|
|
harness.source,
|
|
mime_type="video/mp4",
|
|
on_progress=harness.report,
|
|
on_processing=harness.processing,
|
|
)
|
|
|
|
assert harness.events == ["ossCredentials", "sdk-start", "sdk-done", "processing", "ossConfirm"]
|
|
assert json.loads(harness.api_requests[0].content) == {
|
|
"type": "video", "name": harness.source.name
|
|
}
|
|
assert len(harness.sdk_calls) == 1
|
|
config = harness.configs[0]
|
|
call = harness.sdk_calls[0]
|
|
assert config._secret_id == "synthetic-temporary-id"
|
|
assert config._secret_key == "synthetic-temporary-secret"
|
|
assert config._token == "synthetic-temporary-token"
|
|
assert config._region == REGION
|
|
assert config._scheme == "https"
|
|
assert config._timeout == 120
|
|
assert config._verify_ssl is True
|
|
assert config._allow_redirects is False
|
|
assert config._domain is None
|
|
assert config._ip is None
|
|
assert config._port is None
|
|
assert config.uri(bucket=BUCKET, path=call["Key"]).startswith(
|
|
f"https://{BUCKET}.cos.{REGION}.myqcloud.com/"
|
|
)
|
|
assert call["Bucket"] == BUCKET
|
|
assert call["LocalFilePath"] == str(harness.source.resolve())
|
|
assert call["PartSize"] == 1
|
|
assert call["MAXThread"] == 3
|
|
assert call["ContentType"] == "video/mp4"
|
|
assert set(call) == {
|
|
"Bucket", "Key", "LocalFilePath", "PartSize", "MAXThread", "ContentType", "progress_callback"
|
|
}
|
|
assert re.fullmatch(re.escape(PREFIX) + r"[0-9]+-[0-9a-f]{32}\.mp4", call["Key"])
|
|
assert json.loads(harness.api_requests[1].content) == {
|
|
"type": "video", "key": call["Key"], "name": harness.source.name,
|
|
"size": size, "content_type": "video/mp4", "cid": 0,
|
|
}
|
|
assert result["id"] == 19
|
|
assert result["uri"].endswith("/" + call["Key"])
|
|
assert harness.progress[0] == (0, size)
|
|
assert harness.progress[-1] == (size - 1, size)
|
|
assert all(0 <= sent < total == size for sent, total in harness.progress)
|
|
|
|
|
|
def test_fixed_object_key_is_used_without_expanding_its_scope(harness: _Harness) -> None:
|
|
harness.credentials["object_key"] = PREFIX + "session/fixed.mp4"
|
|
with harness.client() as client:
|
|
upload_video_to_cos(client, harness.source, mime_type="video/mp4")
|
|
assert harness.sdk_calls[0]["Key"] == PREFIX + "session/fixed.mp4"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field_name", "invalid"),
|
|
[
|
|
("provider", "local"),
|
|
("fallback", True),
|
|
("fallback", "false"),
|
|
("bucket", "bucket.example.test/escape-1234567890"),
|
|
("bucket", "bucket@attacker-1234567890"),
|
|
("bucket", ""),
|
|
("region", "ap-guangzhou.attacker.test"),
|
|
("region", "https://attacker.test"),
|
|
("key_prefix", "uploads/voice/20261105/"),
|
|
("key_prefix", "uploads/video/20261105/../"),
|
|
("key_prefix", "uploads/video/20261399/"),
|
|
("key_prefix", "uploads/video/20261105"),
|
|
("object_key", "uploads/video/20261106/other.mp4"),
|
|
("object_key", PREFIX + "../other.mp4"),
|
|
("object_key", PREFIX + "%2e%2e/other.mp4"),
|
|
("object_key", PREFIX + "nested\\other.mp4"),
|
|
("object_key", PREFIX + "other.exe"),
|
|
("object_key", []),
|
|
("max_size", 0),
|
|
("max_size", True),
|
|
("start_time", 9999999999),
|
|
("expired_time", 1),
|
|
("expired_time", False),
|
|
("credentials", None),
|
|
("credentials", {"tmpSecretId": "id", "tmpSecretKey": "secret"}),
|
|
("credentials", {"tmpSecretId": "id", "tmpSecretKey": "", "sessionToken": "token"}),
|
|
("credentials", {"tmpSecretId": "id", "tmpSecretKey": "secret", "sessionToken": "bad\r\n"}),
|
|
],
|
|
)
|
|
def test_invalid_credentials_fail_before_any_cos_upload(
|
|
harness: _Harness, field_name: str, invalid: Any
|
|
) -> None:
|
|
harness.credentials[field_name] = invalid
|
|
with harness.client() as client, pytest.raises(ApiProtocolError):
|
|
upload_video_to_cos(client, harness.source, mime_type="video/mp4")
|
|
assert harness.sdk_calls == []
|
|
assert harness.events == ["ossCredentials"]
|
|
|
|
|
|
@pytest.mark.parametrize("response", [None, [], "invalid"])
|
|
def test_non_mapping_credentials_are_rejected(harness: _Harness, response: Any) -> None:
|
|
harness.credentials = response
|
|
with harness.client() as client, pytest.raises(ApiProtocolError):
|
|
upload_video_to_cos(client, harness.source, mime_type="video/mp4")
|
|
assert harness.sdk_calls == []
|
|
|
|
|
|
def test_server_size_limit_is_enforced_before_cos_upload(harness: _Harness) -> None:
|
|
harness.credentials["max_size"] = harness.source.stat().st_size - 1
|
|
with harness.client() as client, pytest.raises(ValueError, match="上限"):
|
|
upload_video_to_cos(client, harness.source, mime_type="video/mp4")
|
|
assert harness.sdk_calls == []
|
|
assert harness.events == ["ossCredentials"]
|
|
|
|
|
|
@pytest.mark.parametrize("condition", ["missing", "empty", "oversized", "wrong-extension", "bad-mime"])
|
|
def test_invalid_local_video_fails_before_request(harness: _Harness, condition: str) -> None:
|
|
mime = "video/mp4"
|
|
if condition == "missing":
|
|
harness.source.unlink()
|
|
elif condition == "empty":
|
|
harness.source.write_bytes(b"")
|
|
elif condition == "oversized":
|
|
with harness.source.open("wb") as stream:
|
|
stream.truncate(2 * 1024 * 1024 * 1024 + 1)
|
|
elif condition == "wrong-extension":
|
|
renamed = harness.source.with_suffix(".exe")
|
|
harness.source.rename(renamed)
|
|
harness.source = renamed
|
|
else:
|
|
mime = "video/mp4\r\nx-injected: value"
|
|
with harness.client() as client, pytest.raises(ValueError):
|
|
upload_video_to_cos(client, harness.source, mime_type=mime)
|
|
assert harness.api_requests == []
|
|
assert harness.sdk_calls == []
|
|
|
|
|
|
def test_cos_error_is_sanitized_and_never_confirmed(
|
|
harness: _Harness, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
harness.sdk_error = CosClientError("synthetic-temporary-secret synthetic-temporary-token")
|
|
harness.log_sensitive_sdk_data = True
|
|
caplog.set_level(logging.DEBUG)
|
|
with harness.client() as client, pytest.raises(ApiTransportError) as caught:
|
|
upload_video_to_cos(
|
|
client, harness.source, mime_type="video/mp4",
|
|
on_progress=harness.report, on_processing=harness.processing,
|
|
)
|
|
rendered = "".join(traceback.format_exception(caught.value))
|
|
assert "synthetic-temporary-secret" not in rendered + caplog.text
|
|
assert "synthetic-temporary-token" not in rendered + caplog.text
|
|
assert caught.value.data == {"provider": "cos", "operation": "upload"}
|
|
assert caught.value.__suppress_context__ is True
|
|
assert harness.events == ["ossCredentials", "sdk-start"]
|
|
assert harness.progress == [(0, harness.source.stat().st_size)]
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", ["ossCredentials", "ossConfirm"])
|
|
def test_api_business_errors_keep_their_type_and_are_not_retried(
|
|
harness: _Harness, endpoint: str
|
|
) -> None:
|
|
harness.api_failure = endpoint
|
|
with harness.client() as client, pytest.raises(ApiBusinessError, match="synthetic API rejection"):
|
|
upload_video_to_cos(client, harness.source, mime_type="video/mp4", on_progress=harness.report)
|
|
assert sum(request.url.path.endswith(endpoint) for request in harness.api_requests) == 1
|
|
assert len(harness.sdk_calls) == (0 if endpoint == "ossCredentials" else 1)
|
|
assert all(sent < total for sent, total in harness.progress)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"invalid",
|
|
[
|
|
None,
|
|
[],
|
|
{},
|
|
{"id": 0},
|
|
{"id": True},
|
|
{"id": 19},
|
|
{"id": 19, "uri": "/uploads/video/local.mp4"},
|
|
{"id": 19, "uri": "file:///tmp/local.mp4"},
|
|
{"id": 19, "uri": "https://cdn.example.test/different.mp4"},
|
|
{"id": 19, "uri": "https://cdn.example.test:bad/" + PREFIX + "fixed.mp4"},
|
|
{"id": 19, "uri": "https://user:pass@cdn.example.test/" + PREFIX + "fixed.mp4"},
|
|
],
|
|
)
|
|
def test_invalid_confirmation_never_reports_completion(harness: _Harness, invalid: Any) -> None:
|
|
harness.credentials["object_key"] = PREFIX + "fixed.mp4"
|
|
harness.confirmation = invalid
|
|
with harness.client() as client, pytest.raises(ApiProtocolError):
|
|
upload_video_to_cos(client, harness.source, mime_type="video/mp4", on_progress=harness.report)
|
|
assert len(harness.sdk_calls) == 1
|
|
assert harness.events[-1] == "ossConfirm"
|
|
assert all(sent < total for sent, total in harness.progress)
|
|
|
|
|
|
def test_disposed_ui_callbacks_do_not_interrupt_upload_or_confirmation(harness: _Harness) -> None:
|
|
processing = []
|
|
|
|
def broken_progress(_sent: int, _total: int) -> None:
|
|
raise RuntimeError("synthetic disposed progress receiver")
|
|
|
|
def broken_processing() -> None:
|
|
processing.append(True)
|
|
raise RuntimeError("synthetic disposed processing receiver")
|
|
|
|
with harness.client() as client:
|
|
result = upload_video_to_cos(
|
|
client, harness.source, mime_type="video/mp4",
|
|
on_progress=broken_progress, on_processing=broken_processing,
|
|
)
|
|
assert result["id"] == 19
|
|
assert processing == [True]
|
|
assert harness.events == ["ossCredentials", "sdk-start", "sdk-done", "ossConfirm"]
|