HIM-public/ebay_client
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add media upload convenience methods and example workflows for images, documents, and videos

Browse source
This commit is contained in:
claudi 2026-04-07 10:29:42 +02:00
parent ee539e8189
commit 937cd86c8b
4 changed files with 325 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

160
tests/test_public_wrappers.py
View file

@ -16,6 +16,7 @@ from ebay_client.generated.inventory.models import InventoryItemWithSkuLocaleGro
from ebay_client.generated.media.models import (
CreateDocumentFromUrlRequest,
CreateDocumentRequest,
CreateDocumentResponse,
CreateImageFromUrlRequest,
CreateVideoRequest,
DocumentResponse,
@ -562,4 +563,161 @@ def test_media_wait_for_document_raises_on_terminal_failure(monkeypatch) -> None
except ValueError as exc:
assert "REJECTED" in str(exc)
else:
raise AssertionError("Expected wait_for_document to raise on terminal failure status")
raise AssertionError("Expected wait_for_document to raise on terminal failure status")
def test_media_create_upload_and_wait_video_orchestrates_flow(monkeypatch) -> None:
client = MediaClient(build_transport())
calls: list[tuple[str, object]] = []
monkeypatch.setattr(
client,
"create_video",
lambda payload: calls.append(("create_video", payload)) or CreatedMediaResource(resource_id="VIDEO-9"),
)
monkeypatch.setattr(
client,
"upload_video",
lambda video_id, **kwargs: calls.append(("upload_video", {"video_id": video_id, **kwargs})),
)
monkeypatch.setattr(
client,
"wait_for_video",
lambda video_id, **kwargs: calls.append(("wait_for_video", {"video_id": video_id, **kwargs}))
or Video(videoId=video_id, status="LIVE"),
)
result = client.create_upload_and_wait_video(
CreateVideoRequest(title="Demo", size=4, classification=["ITEM"]),
content=b"demo",
poll_interval_seconds=0.0,
)
assert result.videoId == "VIDEO-9"
assert calls[0][0] == "create_video"
assert calls[1] == (
"upload_video",
{"video_id": "VIDEO-9", "content": b"demo", "content_length": 4, "content_range": None},
)
assert calls[2] == (
"wait_for_video",
{"video_id": "VIDEO-9", "timeout_seconds": 30.0, "poll_interval_seconds": 0.0},
)
def test_media_create_upload_and_wait_document_orchestrates_flow(monkeypatch) -> None:
client = MediaClient(build_transport())
calls: list[tuple[str, object]] = []
monkeypatch.setattr(
client,
"create_document",
lambda payload: calls.append(("create_document", payload))
or CreateDocumentResponse(documentId="DOC-9", documentStatus="PENDING_UPLOAD"),
)
monkeypatch.setattr(
client,
"upload_document",
lambda document_id, **kwargs: calls.append(("upload_document", {"document_id": document_id, **kwargs}))
or DocumentResponse(documentId=document_id, documentStatus="SUBMITTED"),
)
monkeypatch.setattr(
client,
"wait_for_document",
lambda document_id, **kwargs: calls.append(("wait_for_document", {"document_id": document_id, **kwargs}))
or DocumentResponse(documentId=document_id, documentStatus="ACCEPTED"),
)
result = client.create_upload_and_wait_document(
CreateDocumentRequest(documentType="USER_GUIDE_OR_MANUAL", languages=["en-US"]),
file_name="guide.pdf",
content=b"%PDF-1.7",
content_type="application/pdf",
poll_interval_seconds=0.0,
)
assert result.documentStatus == "ACCEPTED"
assert calls[0][0] == "create_document"
assert calls[1] == (
"upload_document",
{
"document_id": "DOC-9",
"file_name": "guide.pdf",
"content": b"%PDF-1.7",
"content_type": "application/pdf",
},
)
assert calls[2] == (
"wait_for_document",
{"document_id": "DOC-9", "timeout_seconds": 30.0, "poll_interval_seconds": 0.0},
)
def test_media_create_document_from_url_and_wait_orchestrates_flow(monkeypatch) -> None:
client = MediaClient(build_transport())
calls: list[tuple[str, object]] = []
monkeypatch.setattr(
client,
"create_document_from_url",
lambda payload: calls.append(("create_document_from_url", payload))
or CreateDocumentResponse(documentId="DOC-10", documentStatus="SUBMITTED"),
)
monkeypatch.setattr(
client,
"wait_for_document",
lambda document_id, **kwargs: calls.append(("wait_for_document", {"document_id": document_id, **kwargs}))
or DocumentResponse(documentId=document_id, documentStatus="ACCEPTED"),
)
result = client.create_document_from_url_and_wait(
CreateDocumentFromUrlRequest(
documentType="USER_GUIDE_OR_MANUAL",
documentUrl="https://example.test/guide.pdf",
languages=["en-US"],
),
poll_interval_seconds=0.0,
)
assert result.documentStatus == "ACCEPTED"
assert calls[0][0] == "create_document_from_url"
assert calls[1] == (
"wait_for_document",
{"document_id": "DOC-10", "timeout_seconds": 30.0, "poll_interval_seconds": 0.0},
)
def test_media_convenience_methods_raise_when_required_ids_are_missing(monkeypatch) -> None:
client = MediaClient(build_transport())
monkeypatch.setattr(client, "create_video", lambda payload: CreatedMediaResource(resource_id=None))
monkeypatch.setattr(client, "create_document", lambda payload: CreateDocumentResponse(documentId=None))
monkeypatch.setattr(client, "create_document_from_url", lambda payload: CreateDocumentResponse(documentId=None))
for action in (
lambda: client.create_upload_and_wait_video(
CreateVideoRequest(title="Demo", size=4, classification=["ITEM"]),
content=b"demo",
poll_interval_seconds=0.0,
),
lambda: client.create_upload_and_wait_document(
CreateDocumentRequest(documentType="USER_GUIDE_OR_MANUAL", languages=["en-US"]),
file_name="guide.pdf",
content=b"%PDF-1.7",
poll_interval_seconds=0.0,
),
lambda: client.create_document_from_url_and_wait(
CreateDocumentFromUrlRequest(
documentType="USER_GUIDE_OR_MANUAL",
documentUrl="https://example.test/guide.pdf",
languages=["en-US"],
),
poll_interval_seconds=0.0,
),
):
try:
action()
except RuntimeError:
pass
else:
raise AssertionError("Expected convenience method to raise when eBay omits the required identifier")