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

Add payment dispute management methods and tests in FulfillmentClient

Browse source
This commit is contained in:
claudi 2026-04-07 11:07:17 +02:00
parent 1f06ec9e44
commit f30b31ec00
3 changed files with 411 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

5
ebay_client/core/http/transport.py
View file

@ -45,7 +45,10 @@ class ApiTransport:
filtered_params = None
if params is not None:
filtered_params = {key: value for key, value in params.items() if value is not None}
url = f"{self.base_url}{path}"
if path.startswith(("http://", "https://")):
url = path
else:
url = f"{self.base_url}{path}"
try:
with httpx.Client(timeout=self.timeout_seconds) as client:
response = client.request(

215
ebay_client/fulfillment/client.py
View file

@ -1,22 +1,77 @@
from __future__ import annotations
from urllib.parse import urlparse
from pydantic import BaseModel
from ebay_client.core.http.transport import ApiTransport
from ebay_client.generated.fulfillment.models import (
AcceptPaymentDisputeRequest,
AddEvidencePaymentDisputeRequest,
AddEvidencePaymentDisputeResponse,
ContestPaymentDisputeRequest,
DisputeSummaryResponse,
FileEvidence,
IssueRefundRequest,
Order,
OrderSearchPagedCollection,
PaymentDispute,
PaymentDisputeActivityHistory,
Refund,
ShippingFulfillment,
ShippingFulfillmentDetails,
ShippingFulfillmentPagedCollection,
UpdateEvidencePaymentDisputeRequest,
)
FULFILLMENT_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.fulfillment"
FULFILLMENT_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.fulfillment.readonly"
FULFILLMENT_READ_SCOPE_OPTIONS = [[FULFILLMENT_READ_SCOPE], [FULFILLMENT_SCOPE]]
FINANCES_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.finances"
PAYMENT_DISPUTE_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.payment.dispute"
PAYMENT_DISPUTE_BASE_URL = "https://apiz.ebay.com/sell/fulfillment/v1"
class CreatedShippingFulfillment(BaseModel):
location: str | None = None
fulfillment_id: str | None = None
class EvidenceFileDownload(BaseModel):
content: bytes
content_disposition: str | None = None
file_name: str | None = None
def extract_fulfillment_id(location: str | None) -> str | None:
if not location:
return None
path = urlparse(location).path.rstrip("/")
if not path:
return None
_, _, resource_id = path.rpartition("/")
return resource_id or None
def _extract_file_name(content_disposition: str | None) -> str | None:
if not content_disposition or "filename=" not in content_disposition:
return None
value = content_disposition.split("filename=", 1)[1].strip()
if value.startswith('"') and value.endswith('"'):
return value[1:-1]
return value
class FulfillmentClient:
def __init__(self, transport: ApiTransport) -> None:
self.transport = transport
def _dispute_path(self, path: str) -> str:
return f"{PAYMENT_DISPUTE_BASE_URL}{path}"
def get_order(self, order_id: str) -> Order:
return self.transport.request_model(
Order,
@ -34,6 +89,16 @@ class FulfillmentClient:
params={"limit": limit, "offset": offset},
)
def issue_refund(self, order_id: str, payload: IssueRefundRequest) -> Refund:
return self.transport.request_model(
Refund,
"POST",
f"/sell/fulfillment/v1/order/{order_id}/issue_refund",
scopes=[FINANCES_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def get_shipping_fulfillments(self, order_id: str) -> ShippingFulfillmentPagedCollection:
return self.transport.request_model(
ShippingFulfillmentPagedCollection,
@ -49,3 +114,153 @@ class FulfillmentClient:
f"/sell/fulfillment/v1/order/{order_id}/shipping_fulfillment/{fulfillment_id}",
scope_options=FULFILLMENT_READ_SCOPE_OPTIONS,
)
def create_shipping_fulfillment(
self,
order_id: str,
payload: ShippingFulfillmentDetails,
) -> CreatedShippingFulfillment:
response = self.transport.request(
"POST",
f"/sell/fulfillment/v1/order/{order_id}/shipping_fulfillment",
scopes=[FULFILLMENT_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
location = response.headers.get("Location")
return CreatedShippingFulfillment(location=location, fulfillment_id=extract_fulfillment_id(location))
def get_payment_dispute(self, payment_dispute_id: str) -> PaymentDispute:
return self.transport.request_model(
PaymentDispute,
"GET",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}"),
scopes=[PAYMENT_DISPUTE_SCOPE],
)
def fetch_evidence_content(
self,
payment_dispute_id: str,
*,
evidence_id: str,
file_id: str,
) -> EvidenceFileDownload:
response = self.transport.request(
"GET",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/fetch_evidence_content"),
scopes=[PAYMENT_DISPUTE_SCOPE],
params={"evidence_id": evidence_id, "file_id": file_id},
)
content_disposition = response.headers.get("content-disposition")
return EvidenceFileDownload(
content=response.content,
content_disposition=content_disposition,
file_name=_extract_file_name(content_disposition),
)
def get_payment_dispute_activities(self, payment_dispute_id: str) -> PaymentDisputeActivityHistory:
return self.transport.request_model(
PaymentDisputeActivityHistory,
"GET",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/activity"),
scopes=[PAYMENT_DISPUTE_SCOPE],
)
def get_payment_dispute_summaries(
self,
*,
order_id: str | None = None,
buyer_username: str | None = None,
open_date_from: str | None = None,
open_date_to: str | None = None,
payment_dispute_status: list[str] | None = None,
limit: int | None = None,
offset: int | None = None,
) -> DisputeSummaryResponse:
params: dict[str, object] = {
"order_id": order_id,
"buyer_username": buyer_username,
"open_date_from": open_date_from,
"open_date_to": open_date_to,
"limit": limit,
"offset": offset,
}
if payment_dispute_status is not None:
params["payment_dispute_status"] = payment_dispute_status
return self.transport.request_model(
DisputeSummaryResponse,
"GET",
self._dispute_path("/payment_dispute_summary"),
scopes=[PAYMENT_DISPUTE_SCOPE],
params=params,
)
def contest_payment_dispute(
self,
payment_dispute_id: str,
payload: ContestPaymentDisputeRequest,
) -> None:
self.transport.request_json(
"POST",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/contest"),
scopes=[PAYMENT_DISPUTE_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def accept_payment_dispute(
self,
payment_dispute_id: str,
payload: AcceptPaymentDisputeRequest,
) -> None:
self.transport.request_json(
"POST",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/accept"),
scopes=[PAYMENT_DISPUTE_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def upload_evidence_file(
self,
payment_dispute_id: str,
*,
file_name: str,
content: bytes,
content_type: str = "image/jpeg",
) -> FileEvidence:
return self.transport.request_model(
FileEvidence,
"POST",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/upload_evidence_file"),
scopes=[PAYMENT_DISPUTE_SCOPE],
files={"file": (file_name, content, content_type)},
)
def add_evidence(
self,
payment_dispute_id: str,
payload: AddEvidencePaymentDisputeRequest,
) -> AddEvidencePaymentDisputeResponse:
return self.transport.request_model(
AddEvidencePaymentDisputeResponse,
"POST",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/add_evidence"),
scopes=[PAYMENT_DISPUTE_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def update_evidence(
self,
payment_dispute_id: str,
payload: UpdateEvidencePaymentDisputeRequest,
) -> None:
self.transport.request_json(
"POST",
self._dispute_path(f"/payment_dispute/{payment_dispute_id}/update_evidence"),
scopes=[PAYMENT_DISPUTE_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)

194
tests/test_public_wrappers.py
View file

@ -8,7 +8,7 @@ from ebay_client.account.client import AccountClient
from ebay_client.core.auth.models import OAuthToken
from ebay_client.core.http.transport import ApiTransport
from ebay_client.feed.client import CreatedFeedResource, FeedClient, FeedFileDownload
from ebay_client.fulfillment.client import FulfillmentClient
from ebay_client.fulfillment.client import CreatedShippingFulfillment, EvidenceFileDownload, FulfillmentClient
from ebay_client.generated.account.models import (
FulfillmentPolicy,
FulfillmentPolicyRequest,
@ -26,7 +26,23 @@ from ebay_client.generated.feed.models import (
UpdateUserScheduleRequest,
UserScheduleResponse,
)
from ebay_client.generated.fulfillment.models import Order
from ebay_client.generated.fulfillment.models import (
AcceptPaymentDisputeRequest,
AddEvidencePaymentDisputeRequest,
AddEvidencePaymentDisputeResponse,
FileEvidence,
IssueRefundRequest,
Order,
OrderLineItems,
PaymentDispute,
PaymentDisputeActivityHistory,
Refund,
ShippingFulfillmentDetails,
LineItemReference,
ContestPaymentDisputeRequest,
DisputeSummaryResponse,
UpdateEvidencePaymentDisputeRequest,
)
from ebay_client.generated.inventory.models import (
Address,
BaseResponse,
@ -517,6 +533,180 @@ def test_fulfillment_wrapper_returns_order_model(httpx_mock: HTTPXMock) -> None:
assert result.orderId == "ORDER-1"
def test_fulfillment_wrapper_supports_refund_and_shipping_creation(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/sell/fulfillment/v1/order/ORDER-1/issue_refund",
json={"refundId": "REFUND-1", "refundStatus": "PENDING"},
)
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/sell/fulfillment/v1/order/ORDER-1/shipping_fulfillment",
status_code=201,
headers={
"Location": "https://api.ebay.com/sell/fulfillment/v1/order/ORDER-1/shipping_fulfillment/FULFILL-1"
},
json={},
)
client = FulfillmentClient(build_transport())
refund = client.issue_refund(
"ORDER-1",
IssueRefundRequest(reasonForRefund="BUYER_CANCELLED", comment="requested", orderLevelRefundAmount={"currency": "USD", "value": "10.00"}),
)
fulfillment = client.create_shipping_fulfillment(
"ORDER-1",
ShippingFulfillmentDetails(
lineItems=[LineItemReference(lineItemId="LINE-1", quantity=1)],
shippingCarrierCode="USPS",
trackingNumber="TRACK123",
),
)
assert isinstance(refund, Refund)
assert refund.refundId == "REFUND-1"
assert isinstance(fulfillment, CreatedShippingFulfillment)
assert fulfillment.fulfillment_id == "FULFILL-1"
refund_request = httpx_mock.get_requests()[0]
refund_body = json.loads(refund_request.content.decode("utf-8"))
assert refund_body["reasonForRefund"] == "BUYER_CANCELLED"
assert refund_request.headers["Authorization"] == "Bearer test-token"
shipping_request = httpx_mock.get_requests()[1]
shipping_body = json.loads(shipping_request.content.decode("utf-8"))
assert shipping_body["shippingCarrierCode"] == "USPS"
assert shipping_body["trackingNumber"] == "TRACK123"
def test_fulfillment_wrapper_supports_payment_dispute_workflow(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1",
json={"paymentDisputeId": "DISPUTE-1", "revision": 3},
)
httpx_mock.add_response(
method="GET",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/activity",
json={"activities": []},
)
httpx_mock.add_response(
method="GET",
url=(
"https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute_summary?buyer_username=buyer1"
"&payment_dispute_status=OPEN&payment_dispute_status=ACTION_NEEDED&limit=25&offset=0"
),
json={"paymentDisputeSummaries": [], "total": 0},
)
httpx_mock.add_response(
method="POST",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/contest",
status_code=204,
)
httpx_mock.add_response(
method="POST",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/accept",
status_code=204,
)
httpx_mock.add_response(
method="POST",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/upload_evidence_file",
json={"fileId": "FILE-1", "fileName": "label.jpg"},
)
httpx_mock.add_response(
method="POST",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/add_evidence",
json={"evidenceId": "EVID-1"},
)
httpx_mock.add_response(
method="POST",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/update_evidence",
status_code=204,
)
httpx_mock.add_response(
method="GET",
url="https://apiz.ebay.com/sell/fulfillment/v1/payment_dispute/DISPUTE-1/fetch_evidence_content?evidence_id=EVID-1&file_id=FILE-1",
status_code=200,
headers={"content-disposition": 'attachment; filename="label.jpg"'},
content=b"binary-evidence",
)
client = FulfillmentClient(build_transport())
dispute = client.get_payment_dispute("DISPUTE-1")
activities = client.get_payment_dispute_activities("DISPUTE-1")
summaries = client.get_payment_dispute_summaries(
buyer_username="buyer1",
payment_dispute_status=["OPEN", "ACTION_NEEDED"],
limit=25,
offset=0,
)
client.contest_payment_dispute("DISPUTE-1", ContestPaymentDisputeRequest(revision=3, note="tracking attached"))
client.accept_payment_dispute("DISPUTE-1", AcceptPaymentDisputeRequest(revision=3))
file_evidence = client.upload_evidence_file(
"DISPUTE-1",
file_name="label.jpg",
content=b"jpg-bytes",
content_type="image/jpeg",
)
evidence_response = client.add_evidence(
"DISPUTE-1",
AddEvidencePaymentDisputeRequest(
evidenceType="PROOF_OF_DELIVERY",
files=[FileEvidence(fileId="FILE-1")],
lineItems=[OrderLineItems(lineItemId="LINE-1", itemId="ITEM-1")],
),
)
client.update_evidence(
"DISPUTE-1",
UpdateEvidencePaymentDisputeRequest(
evidenceId="EVID-1",
evidenceType="PROOF_OF_DELIVERY",
files=[FileEvidence(fileId="FILE-1")],
lineItems=[OrderLineItems(lineItemId="LINE-1", itemId="ITEM-1")],
),
)
evidence_file = client.fetch_evidence_content("DISPUTE-1", evidence_id="EVID-1", file_id="FILE-1")
assert isinstance(dispute, PaymentDispute)
assert isinstance(activities, PaymentDisputeActivityHistory)
assert isinstance(summaries, DisputeSummaryResponse)
assert summaries.total == 0
assert isinstance(file_evidence, FileEvidence)
assert file_evidence.fileId == "FILE-1"
assert isinstance(evidence_response, AddEvidencePaymentDisputeResponse)
assert evidence_response.evidenceId == "EVID-1"
assert isinstance(evidence_file, EvidenceFileDownload)
assert evidence_file.file_name == "label.jpg"
assert evidence_file.content == b"binary-evidence"
contest_request = httpx_mock.get_requests()[3]
assert contest_request.url.host == "apiz.ebay.com"
contest_body = json.loads(contest_request.content.decode("utf-8"))
assert contest_body["revision"] == 3
upload_request = httpx_mock.get_requests()[5]
assert upload_request.url.host == "apiz.ebay.com"
assert upload_request.headers["Content-Type"].startswith("multipart/form-data;")
def test_fulfillment_wrapper_accepts_readonly_or_full_scope_options_for_shipping_reads(httpx_mock: HTTPXMock) -> None:
oauth_client = RecordingOAuthClient()
transport = ApiTransport(base_url="https://api.ebay.com", oauth_client=oauth_client)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/fulfillment/v1/order/ORDER-1/shipping_fulfillment/FULL-1",
json={"fulfillmentId": "FULL-1"},
)
FulfillmentClient(transport).get_shipping_fulfillment("ORDER-1", "FULL-1")
assert oauth_client.last_scopes is None
assert oauth_client.last_scope_options == [
["https://api.ebay.com/oauth/api_scope/sell.fulfillment.readonly"],
["https://api.ebay.com/oauth/api_scope/sell.fulfillment"],
]
def test_account_wrapper_returns_programs_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",