Add generated TUS protocol canary

.github/workflows/CI.yml

@@ -13,6 +13,8 @@ jobs:
 
     steps:
       - uses: actions/checkout@v6
+        with:
+          lfs: true
 
       - name: Set up Python
         uses: actions/setup-python@v6

examples/api2-devdock-transloadit-assembly-upload/main.py

@@ -0,0 +1,61 @@
+"""Upload to a Transloadit devdock Assembly using tus-py-client.
+
+This example is intentionally checked into the SDK repository. API2 owns the
+scenario JSON and prepares the live Transloadit Assembly; this file only shows
+ordinary tus-py-client usage against the injected TUS endpoint.
+"""
+
+import sys
+from io import BytesIO
+from pathlib import Path
+
+from tusclient import client as tus
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from api2devdock import (
+    fail,
+    load_scenario,
+    scenario_bytes,
+    tus_url,
+    upload_metadata,
+    write_result,
+)
+
+
+def upload_with_tus(scenario, create_response):
+    upload_config = scenario["upload"]
+    endpoint_url = tus_url(upload_config, scenario, create_response)
+    content = scenario_bytes(upload_config)
+    if upload_config["chunkSize"] != "full-file":
+        fail("unsupported chunk size policy {!r}".format(upload_config["chunkSize"]))
+
+    uploader = tus.TusClient(endpoint_url).uploader(
+        file_stream=BytesIO(content),
+        chunk_size=len(content),
+        metadata=upload_metadata(upload_config, scenario, create_response),
+        retries=upload_config["retries"],
+    )
+    uploader.upload()
+
+    if not uploader.url:
+        fail("TUS upload did not expose an upload URL")
+    if uploader.offset != len(content):
+        fail("TUS upload offset {}, expected {}".format(uploader.offset, len(content)))
+
+    return uploader.url
+
+
+def main():
+    scenario = load_scenario(Path(__file__).with_name("api2-scenario.json"))
+    create_response = scenario["prepared"]["createResponse"]
+    upload_url = upload_with_tus(scenario, create_response)
+    write_result({"uploadUrl": upload_url})
+    print(
+        "Python TUS SDK devdock scenario {} uploaded to {}".format(
+            scenario["scenarioId"], upload_url
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

examples/api2-devdock-tus-abort-upload/main.py

@@ -0,0 +1,155 @@
+"""Abort a TUS upload against the API2 devdock conformance server."""
+
+import sys
+from io import BytesIO
+from pathlib import Path
+
+from tusclient import client as tus
+from tusclient.exceptions import TusUploadAborted
+from tusclient.fingerprint.interface import Fingerprint
+from tusclient.storage.interface import Storage
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from api2devdock import (
+    TusConformancePlanServer,
+    bool_value,
+    conformance_input_options,
+    conformance_input_source_bytes,
+    fail,
+    load_scenario,
+    object_value,
+    scenario_id,
+    string_value,
+    write_result,
+)
+
+
+class MemoryStorage(Storage):
+    def __init__(self):
+        self.values = {}
+
+    def get_item(self, key):
+        return self.values.get(key)
+
+    def set_item(self, key, value):
+        self.values[key] = value
+
+    def remove_item(self, key):
+        self.values.pop(key, None)
+
+
+class FixedFingerprint(Fingerprint):
+    def __init__(self, fingerprint):
+        self.fingerprint = fingerprint
+
+    def get_fingerprint(self, fs):
+        return self.fingerprint
+
+
+def upload_and_abort(conformance_scenario):
+    input_options = conformance_input_options(conformance_scenario)
+    content = conformance_input_source_bytes(conformance_scenario)
+    endpoint_url = string_value(input_options["endpointUrl"], "endpointUrl")
+    headers = object_value(input_options.get("headers", {}), "headers")
+    metadata = object_value(input_options["metadata"], "metadata")
+    override_patch_method = bool_value(
+        input_options.get("overridePatchMethod", False),
+        "overridePatchMethod",
+    )
+    runtime_setup = object_value(
+        conformance_scenario["runtimeSetup"],
+        "conformanceScenario.runtimeSetup",
+    )
+    abort_setup = object_value(
+        runtime_setup["abort"],
+        "conformanceScenario.runtimeSetup.abort",
+    )
+    terminate_upload_on_abort = bool_value(
+        abort_setup["terminateUpload"],
+        "conformanceScenario.runtimeSetup.abort.terminateUpload",
+    )
+
+    client_ref = {}
+    uploader_ref = {}
+
+    def on_abort_request(event):
+        active_client = client_ref.get("client")
+        active_uploader = uploader_ref.get("uploader")
+        if active_client is None:
+            fail("abort request observed before client was initialized")
+        active_client.abort_upload(active_uploader, False)
+
+    with TusConformancePlanServer(
+        conformance_scenario,
+        endpoint_url,
+        on_abort_request=on_abort_request,
+    ) as conformance_server:
+        client = tus.TusClient(conformance_server.endpoint_url(), headers=headers)
+        client_ref["client"] = client
+
+        storage = None
+        fingerprint = input_options.get("fingerprint")
+        uploader_options = {}
+        if fingerprint is not None:
+            storage = MemoryStorage()
+            uploader_options.update(
+                {
+                    "fingerprinter": FixedFingerprint(
+                        string_value(fingerprint, "fingerprint"),
+                    ),
+                    "store_url": True,
+                    "url_storage": storage,
+                }
+            )
+
+        uploader = client.uploader(
+            file_stream=BytesIO(content),
+            chunk_size=len(content),
+            metadata=metadata,
+            override_patch_method=override_patch_method,
+            **uploader_options,
+        )
+        uploader_ref["uploader"] = uploader
+
+        try:
+            uploader.upload()
+        except TusUploadAborted:
+            pass
+        else:
+            fail("abort scenario completed without TusUploadAborted")
+
+        if terminate_upload_on_abort:
+            if not uploader.url:
+                fail("abort scenario requested termination before upload URL was known")
+            client.abort_upload(uploader, True)
+
+        conformance_server.assert_exhausted()
+        result = conformance_server.result()
+        result["completionKind"] = "aborted"
+        result["errorCalled"] = False
+        result["successCalled"] = False
+        result["uploadUrl"] = (
+            conformance_server.canonical_url(uploader.url) if uploader.url else None
+        )
+        if storage is not None and fingerprint is not None:
+            result["storedUrlAfterAbort"] = storage.get_item(fingerprint)
+        return result
+
+
+def main():
+    scenario = load_scenario(Path(__file__).with_name("api2-scenario.json"))
+    conformance_scenario = object_value(
+        scenario["conformanceScenario"],
+        "conformanceScenario",
+    )
+    result = upload_and_abort(conformance_scenario)
+    write_result(result)
+    print(
+        "Python TUS SDK devdock scenario {} aborted the upload".format(
+            scenario_id(scenario),
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

examples/api2-devdock-tus-creation-with-upload/main.py

@@ -0,0 +1,71 @@
+"""Create a Transloadit devdock TUS upload with bytes in the creation request."""
+
+import sys
+from io import BytesIO
+from pathlib import Path
+
+from tusclient import client as tus
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from api2devdock import (
+    fail,
+    load_scenario,
+    scenario_bytes,
+    scenario_id,
+    tus_url,
+    upload_metadata,
+    write_result,
+)
+
+
+def upload_with_creation_body(scenario, create_response):
+    upload_config = scenario["upload"]
+    content = scenario_bytes(upload_config)
+
+    if upload_config["chunkSize"] != "full-file":
+        fail("unsupported chunk size policy {!r}".format(upload_config["chunkSize"]))
+    if not upload_config["uploadDataDuringCreation"]:
+        fail("scenario does not enable uploadDataDuringCreation")
+
+    uploader = tus.TusClient(
+        tus_url(upload_config, scenario, create_response)
+    ).create_upload_with_data(
+        len(content),
+        file_stream=BytesIO(content),
+        chunk_size=len(content),
+        metadata=upload_metadata(upload_config, scenario, create_response),
+        retries=upload_config["retries"],
+    )
+    uploader.upload()
+
+    if not uploader.url:
+        fail("creation-with-upload did not expose an upload URL")
+    if uploader.offset != len(content):
+        fail(
+            "creation-with-upload accepted {} bytes, expected {}".format(
+                uploader.offset,
+                len(content),
+            )
+        )
+
+    return {
+        "acceptedBytes": uploader.offset,
+        "uploadUrl": uploader.url,
+    }
+
+
+def main():
+    scenario = load_scenario(Path(__file__).with_name("api2-scenario.json"))
+    create_response = scenario["prepared"]["createResponse"]
+    result = upload_with_creation_body(scenario, create_response)
+    write_result(result)
+    print(
+        "Python TUS SDK devdock scenario {} uploaded during creation to {}".format(
+            scenario_id(scenario),
+            result["uploadUrl"],
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

examples/api2-devdock-tus-custom-request-headers/main.py

@@ -0,0 +1,99 @@
+"""Send Transloadit devdock TUS custom request headers."""
+
+import sys
+from io import BytesIO
+from pathlib import Path
+
+from tusclient import client as tus
+from tusclient.request_lifecycle import RequestLifecycleHooks
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from api2devdock import (
+    fail,
+    load_scenario,
+    scenario_bytes,
+    scenario_id,
+    tus_url,
+    upload_headers,
+    upload_metadata,
+    write_result,
+)
+
+
+def record_custom_request_headers(context, expected_headers):
+    observed = {}
+    for header_name, expected_value in expected_headers.items():
+        actual_value = context.headers.get(header_name)
+        if actual_value != expected_value:
+            fail(
+                "custom request header {} expected {!r}, got {!r}".format(
+                    header_name,
+                    expected_value,
+                    actual_value,
+                )
+            )
+        observed[header_name] = actual_value
+    return observed
+
+
+def upload_with_custom_request_headers(scenario, create_response):
+    upload_config = scenario["upload"]
+    expected_headers = upload_headers(scenario)
+    content = scenario_bytes(upload_config)
+    headers_by_method = {}
+    if upload_config["chunkSize"] != "full-file":
+        fail("unsupported chunk size policy {!r}".format(upload_config["chunkSize"]))
+
+    def before_request(context):
+        if context.method in ("POST", "PATCH"):
+            headers_by_method[context.method] = record_custom_request_headers(
+                context,
+                expected_headers,
+            )
+
+    uploader = tus.TusClient(
+        tus_url(upload_config, scenario, create_response),
+        headers=expected_headers,
+        request_hooks=RequestLifecycleHooks(before_request=before_request),
+    ).uploader(
+        file_stream=BytesIO(content),
+        chunk_size=len(content),
+        metadata=upload_metadata(upload_config, scenario, create_response),
+        retries=upload_config["retries"],
+    )
+    uploader.upload()
+
+    if not uploader.url:
+        fail("custom request headers upload did not expose an upload URL")
+    if uploader.offset != len(content):
+        fail(
+            "custom request headers upload offset {}, expected {}".format(
+                uploader.offset,
+                len(content),
+            )
+        )
+    for method in ("POST", "PATCH"):
+        if method not in headers_by_method:
+            fail("custom request headers did not observe {} request".format(method))
+
+    return {
+        "headersByMethod": headers_by_method,
+        "uploadUrl": uploader.url,
+    }
+
+
+def main():
+    scenario = load_scenario(Path(__file__).with_name("api2-scenario.json"))
+    create_response = scenario["prepared"]["createResponse"]
+    result = upload_with_custom_request_headers(scenario, create_response)
+    write_result(result)
+    print(
+        "Python TUS SDK devdock scenario {} sent custom request headers for {}".format(
+            scenario_id(scenario),
+            result["uploadUrl"],
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()