primus.storage.test_graphql_artifact_store module

Behavior specifications for application-authorized artifact transfers.

Feature: GraphQL artifact transfers
Scenario: Upload verified bytes without local AWS credentials

Given an authorized GraphQL ticket executor and a signed HTTPS ticket When a caller uploads bytes matching the declared checksum and size Then the bytes are transferred before metadata is returned and no S3 client is used

Scenario: A signed URL expires during transfer

Given a write ticket whose HTTPS response is RequestExpired When a caller uploads matching bytes Then the store requests exactly one replacement ticket and retries once

Scenario: Authorization or integrity validation fails

Given an unauthorized response or mismatched downloaded content When a caller transfers an artifact Then the operation fails closed without an alternate storage path

class primus.storage.test_graphql_artifact_store.FakeCAHTTPSession(responses: list[FakeResponse | Exception])

Bases: FakeHTTPSession

request(method, url, *, headers, data=None, timeout=None, verify=None)
class primus.storage.test_graphql_artifact_store.FakeExecutor(ticket_batches)

Bases: object

__init__(ticket_batches)
execute(query, variables=None)
class primus.storage.test_graphql_artifact_store.FakeHTTPSession(responses: list[FakeResponse | Exception])

Bases: object

__init__(responses: list[FakeResponse | Exception])
request(method, url, *, headers, data=None, timeout=None)
class primus.storage.test_graphql_artifact_store.FakeResponse(status_code: int, content: bytes = b'', text: str = '')

Bases: object

__init__(status_code: int, content: bytes = b'', text: str = '')
primus.storage.test_graphql_artifact_store.read_request()
primus.storage.test_graphql_artifact_store.test_authorization_failure_does_not_request_a_replacement_ticket()
primus.storage.test_graphql_artifact_store.test_configured_local_ca_bundle_is_used_for_https_verification(tmp_path, monkeypatch)
primus.storage.test_graphql_artifact_store.test_decodes_required_headers_returned_as_graphql_awsjson()
primus.storage.test_graphql_artifact_store.test_download_batch_requests_all_tickets_once_and_verifies_each_payload()
primus.storage.test_graphql_artifact_store.test_download_checksum_failure_fails_closed_without_retries()
primus.storage.test_graphql_artifact_store.test_expired_signed_url_reissues_one_ticket_and_retries_once()
primus.storage.test_graphql_artifact_store.test_missing_configured_local_ca_bundle_fails_closed(monkeypatch)
primus.storage.test_graphql_artifact_store.test_non_retryable_upload_client_status_fails_closed_without_fresh_ticket(status_code)
primus.storage.test_graphql_artifact_store.test_requests_batched_tickets_with_frozen_contract_fields()
primus.storage.test_graphql_artifact_store.test_retryable_download_failure_does_not_request_a_fresh_ticket()
primus.storage.test_graphql_artifact_store.test_retryable_upload_request_exception_gets_one_fresh_ticket_for_identical_bytes()
primus.storage.test_graphql_artifact_store.test_retryable_upload_status_gets_one_fresh_ticket_for_identical_bytes(status_code)
primus.storage.test_graphql_artifact_store.test_second_retryable_upload_failure_fails_closed_without_a_third_ticket()
primus.storage.test_graphql_artifact_store.test_standard_s3_expiry_message_reissues_one_ticket_and_retries_once()
primus.storage.test_graphql_artifact_store.test_ticket_requires_an_https_host()
primus.storage.test_graphql_artifact_store.test_upload_batch_requests_all_tickets_once_before_uploading_each_verified_artifact()
primus.storage.test_graphql_artifact_store.test_upload_omits_only_a_redundant_unsigned_checksum_header_already_bound_in_the_url()
primus.storage.test_graphql_artifact_store.test_upload_preserves_checksum_header_when_query_binding_is_absent_inconsistent_or_signed(url_query, header_value)
primus.storage.test_graphql_artifact_store.test_upload_rejects_mismatched_declared_bytes_before_requesting_ticket()
primus.storage.test_graphql_artifact_store.test_upload_transfers_verified_bytes_before_returning_metadata_without_s3(monkeypatch)
primus.storage.test_graphql_artifact_store.ticket(*, url='https://storage.example/upload', method='PUT')
primus.storage.test_graphql_artifact_store.write_request()