from typing import Any, TypedDict from uuid import UUID import requests from pydantic import UUID4 from tests.integration import config class PresignedURLPart(TypedDict): part_number: int url: str class PresignedURLResponse(TypedDict): part_number: int etag: str class OwsProductStaging: """Client for OWS Product Staging API interactions.""" def __init__(self, *, jwt: str): self.jwt = jwt # Bulk Session CRUD Operations def create_bulk_session( self, *, vendor_uuid: str, expected_status_code: int = 200 ) -> UUID4: response = requests.post( f"{config.QA_BASE_URL}/bulk-session", headers={"Authorization": f"Bearer {self.jwt}"}, json={"vendor_uuid": vendor_uuid}, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() assert "bulk_session_id" in response_body return UUID(response_body["bulk_session_id"]) def create_bulk_session_with_subaccount_id( self, *, subaccount_id: int, expected_status_code: int = 200 ) -> UUID4: response = requests.post( f"{config.QA_BASE_URL}/bulk-session", headers={"Authorization": f"Bearer {self.jwt}"}, json={"subaccount_id": subaccount_id}, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() assert "bulk_session_id" in response_body return UUID(response_body["bulk_session_id"]) def get_metadata_errors( self, *, bulk_session_id: UUID4, expected_status_code: int = 200 ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{bulk_session_id}/metadata-errors.json", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_bulk_session( self, *, bulk_session_id: UUID4, expected_status_code: int = 200 ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() return response_body def get_bulk_session_by_slug( self, *, slug: str, path_suffix: str = "", expected_status_code: int = 307 ) -> str | None: """GET a bulk-session route by slug without following the redirect. ``path_suffix`` (e.g. "/assets") targets a sub-route; returns the Location header. """ response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{slug}{path_suffix}", headers={"Authorization": f"Bearer {self.jwt}"}, allow_redirects=False, ) assert response.status_code == expected_status_code return response.headers.get("Location") def update_bulk_session( self, *, bulk_session_id: UUID4, update_data: dict[str, Any], expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.patch( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, json=update_data, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() # Metadata Operations def start_metadata_upload( self, *, bulk_session_id: UUID4, original_filename: str, expected_status_code: int = 200, ) -> str: response = requests.post( f"{config.QA_BASE_URL}/metadata/upload", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "bulk_session_id": str(bulk_session_id), "original_filename": original_filename, }, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() assert "s3_filename" in response_body assert response_body["bulk_session_id"] == str(bulk_session_id) return response_body["s3_filename"] def get_presigned_urls( self, *, s3_filename: str, parts: int, expected_status_code: int = 200 ) -> list[PresignedURLPart]: response = requests.get( f"{config.QA_BASE_URL}/metadata/upload/{s3_filename}", headers={"Authorization": f"Bearer {self.jwt}"}, params={"parts": parts}, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() return response_body["presigned_urls"] def complete_metadata_upload( self, *, s3_filename: str, parts: list[PresignedURLResponse], expected_status_code: int = 200, ) -> UUID4: """ Completes a multi-part upload for bulk session metadata. Returns the identifier of the created bulk session metadata file. """ response = requests.patch( f"{config.QA_BASE_URL}/metadata/upload/{s3_filename}", headers={"Authorization": f"Bearer {self.jwt}"}, json={"parts": parts}, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() assert "bulk_session_metadata_file_id" in response_body return UUID(response_body["bulk_session_metadata_file_id"]) def get_metadata_file( self, *, bulk_session_id: UUID4, metadata_file_id: UUID4, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/metadata-file/{str(metadata_file_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_metadata_file_link( self, *, bulk_session_id: UUID4, metadata_file_id: UUID4, expected_status_code: int = 200, ) -> str: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/metadata-file/{str(metadata_file_id)}/link", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def finalize_metadata_file( self, *, bulk_session_id: UUID4, bulk_session_metadata_file_id: UUID4, is_classical: bool = False, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.post( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/metadata-file/{str(bulk_session_metadata_file_id)}/finalize", headers={"Authorization": f"Bearer {self.jwt}"}, params={"classical": is_classical}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() # Asset Operations def get_assets( self, *, bulk_session_id: UUID4, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/assets", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_asset_report( self, *, bulk_session_id: UUID4, expected_status_code: int = 200, ) -> dict[str, Any] | None: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/assets/report", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_asset_by_original_filename( self, *, bulk_session_id: UUID4, filename: str, expected_status_code: int = 200, ) -> dict[str, Any] | None: response = requests.get( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/assets/original-filename/{filename}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def start_asset_upload( self, *, bulk_session_id: UUID4, filenames: list[str], parts: int = 1, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.post( f"{config.QA_BASE_URL}/assets/upload", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "bulk_session_id": str(bulk_session_id), "assets": [ {"original_filename": filename, "parts": 1} for filename in filenames ], }, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def update_asset( self, *, bulk_session_id: UUID4, s3_filename: str, update_data: dict[str, Any], expected_status_code: int = 200, ) -> dict[str, Any] | None: response = requests.patch( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/assets/{s3_filename}", headers={"Authorization": f"Bearer {self.jwt}"}, json=update_data, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def delete_asset( self, *, bulk_session_id: UUID4, bulk_session_asset_file_id: UUID4, expected_status_code: int = 204, ) -> None: response = requests.delete( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/asset-file/{str(bulk_session_asset_file_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() def complete_asset_upload( self, *, s3_filename: str, parts: list[PresignedURLResponse], expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.patch( f"{config.QA_BASE_URL}/assets/upload/{s3_filename}", headers={"Authorization": f"Bearer {self.jwt}"}, json={"parts": parts}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def dataload_assets( self, *, bulk_session_id: UUID4, bulk_session_asset_ids: list[UUID4], expected_status_code: int = 200, ) -> list[dict[str, Any]]: response = requests.post( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}/asset-file/dataloader", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "bulk_session_asset_ids": [ str(asset_id) for asset_id in bulk_session_asset_ids ], }, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() # Asset Transfer Job Operations def update_asset_transfer_job( self, *, bulk_session_id: UUID4, bulk_session_asset_cloud_transfer_job_id: UUID4, update_data: dict[str, Any], expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.patch( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}" f"/asset-transfer-job/{str(bulk_session_asset_cloud_transfer_job_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, json=update_data, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def start_google_asset_transfer_job( self, *, bulk_session_id: UUID4, user_id: str = "test-google-user-id", assets: list[dict[str, str]] = [], expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.post( f"{config.QA_BASE_URL}/assets/upload/google", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "bulk_session_id": str(bulk_session_id), "user_id": user_id, "assets": assets, }, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() # Bulk Session Ingestion Operations def ingest_bulk_session( self, *, bulk_session_id: UUID4, assets_required: bool, submit: bool, expected_status_code: int = 200, ) -> UUID4: response = requests.post( f"{config.QA_BASE_URL}/bulk-session/{str(bulk_session_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "assets_required": assets_required, "submit": submit, "send_notifications": False, }, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() assert "bulk_session_ingestion_id" in response_body return UUID(response_body["bulk_session_ingestion_id"]) def get_bulk_session_ingestion( self, *, ingestion_id: UUID4, expected_status_code: int = 200 ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session-ingestion/{str(ingestion_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def update_bulk_session_ingestion( self, *, ingestion_id: UUID4, update_data: dict[str, Any], expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.patch( f"{config.QA_BASE_URL}/bulk-session-ingestion/update/{str(ingestion_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, json=update_data, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def patch_bulk_session_ingestion( self, *, ingestion_id: UUID4, patch_data: dict[str, Any], expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.patch( f"{config.QA_BASE_URL}/bulk-session-ingestion/{str(ingestion_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, json=patch_data, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def upsert_bulk_session_ingestion_product( self, *, ingestion_id: UUID4, product_code: str, execution_arn: str, expected_status_code: int = 200, ) -> dict[str, Any] | None: response = requests.put( f"{config.QA_BASE_URL}/bulk-session-ingestion-product", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "bulk_session_ingestion_id": str(ingestion_id), "product_code": product_code, "execution_arn": execution_arn, }, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_bulk_session_ingestion_products( self, *, ingestion_id: UUID4, expected_status_code: int = 200 ) -> list[dict[str, Any]]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session-ingestion/{str(ingestion_id)}/products", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_latest_ingestion_product_by_product_code( self, *, product_code: str, expected_status_code: int = 200 ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session-ingestion-product/product-code/{product_code}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def get_latest_ingestion_product_by_code_and_session( self, *, product_code: str, bulk_session_id: UUID4, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.get( f"{config.QA_BASE_URL}/bulk-session-ingestion-product/product-code/{product_code}/bulk-session/{str(bulk_session_id)}", headers={"Authorization": f"Bearer {self.jwt}"}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def create_bulk_session_ingestion_track( self, *, bulk_session_ingestion_product_id: UUID4, volume: int = 1, sequence_number: int = 1, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.post( f"{config.QA_BASE_URL}/bulk-session-ingestion-product/{str(bulk_session_ingestion_product_id)}/track", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "volume": volume, "sequence_number": sequence_number, }, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() def upsert_bulk_session_ingestion_track( self, *, bulk_session_ingestion_product_id: UUID4, execution_arn: str, volume: int = 1, sequence_number: int = 1, track_id: int | None = None, isrc: str | None = None, ingestion_status: str | None = None, expected_status_code: int = 200, ) -> dict[str, Any]: response = requests.put( f"{config.QA_BASE_URL}/bulk-session-ingestion-product/{str(bulk_session_ingestion_product_id)}/track", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "execution_arn": execution_arn, "volume": volume, "sequence_number": sequence_number, "track_id": track_id, "isrc": isrc, "ingestion_status": ingestion_status, }, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json() # Template Operations def get_bulk_template( self, params: dict[str, Any] = {}, expected_status_code: int = 200 ) -> str: response = requests.get( f"{config.QA_BASE_URL}/bulk-template", headers={"Authorization": f"Bearer {self.jwt}"}, params=params, ) assert response.status_code == expected_status_code response.raise_for_status() response_body = response.json() download_link = response_body.get("download_link", None) assert download_link return download_link def google_auth_exchange( self, *, bulk_session_id: UUID4, auth_code: str | None = None, user_id: str | None = None, expected_status_code: int = 400, ) -> None: response = requests.put( f"{config.QA_BASE_URL}/assets/upload/google/oauth/exchange", headers={"Authorization": f"Bearer {self.jwt}"}, json={ "bulk_session_id": str(bulk_session_id), "auth_code": auth_code, "user_id": user_id, }, ) assert response.status_code == expected_status_code response.raise_for_status() # Allowed Value Sets def get_validation_enums(self, expected_status_code: int = 200) -> dict[str, Any]: response = requests.post( f"{config.QA_BASE_URL}/validation-enums", headers={"Authorization": f"Bearer {self.jwt}"}, json={}, ) assert response.status_code == expected_status_code response.raise_for_status() return response.json()