| """Actual AWS CLI v2 high-level transfers through a /s3/ endpoint.""" | |
| import json | |
| import os | |
| from pathlib import Path | |
| import subprocess | |
| import tempfile | |
| from uuid import uuid4 | |
| from urllib.parse import urlsplit | |
| import httpx | |
| ROOT = Path(__file__).resolve().parents[1] | |
| WORK = ROOT / "work/buckets-sandbox" | |
| def main(): | |
| config = json.loads((WORK / "config.json").read_text()) | |
| tokens = json.loads((WORK / "credentials.json").read_text()) | |
| endpoint = os.getenv("BUCKETS_E2E_ENDPOINT", config["BUCKETS_PUBLIC_ENDPOINT"]) | |
| native = httpx.Client( | |
| base_url=endpoint.removesuffix("/s3/"), | |
| verify=str(WORK / "localhost.pem"), | |
| trust_env=False, | |
| headers={"Authorization": "Bearer " + tokens["user1_personal"]}, | |
| ) | |
| response = native.post( | |
| "/v1/buckets/s3-credentials", | |
| json={"permissions": ["storage.read", "storage.write", "storage.admin"], "rate_limit_rpm": 600}, | |
| ) | |
| response.raise_for_status() | |
| credential = response.json() | |
| target = urlsplit(endpoint) | |
| endpoint = f"{target.scheme}://{target.netloc}" + urlsplit(credential["endpoint"]).path | |
| env = { | |
| **os.environ, | |
| "AWS_ACCESS_KEY_ID": credential["access_key_id"], | |
| "AWS_SECRET_ACCESS_KEY": credential["secret_access_key"], | |
| "AWS_DEFAULT_REGION": "us-east-1", | |
| "AWS_CA_BUNDLE": str(WORK / "localhost.pem"), | |
| "AWS_EC2_METADATA_DISABLED": "true", | |
| "AWS_PAGER": "", | |
| } | |
| executable = os.getenv("AWS_CLI_BINARY", "aws") | |
| def aws(*args, overrides=None): | |
| result = subprocess.run( | |
| [executable, "--endpoint-url", endpoint, *args], | |
| env={**env, **(overrides or {})}, | |
| capture_output=True, | |
| text=True, | |
| ) | |
| if result.returncode: | |
| raise RuntimeError("AWS CLI operation failed: " + result.stderr[-2000:]) | |
| return result.stdout | |
| name = "aws-e2e-" + uuid4().hex[:12] | |
| try: | |
| with tempfile.TemporaryDirectory(dir=WORK) as temp: | |
| folder = Path(temp) | |
| cfg = folder / "aws-config" | |
| cfg.write_text( | |
| "[default]\nregion=us-east-1\ns3 =\n addressing_style = path\n signature_version = s3v4\n multipart_threshold = 5MB\n multipart_chunksize = 8MB\n" | |
| ) | |
| env["AWS_CONFIG_FILE"] = str(cfg) | |
| source = folder / "model.bin" | |
| source.write_bytes(b"A" * (16 * 1024**2)) | |
| aws("s3api", "create-bucket", "--bucket", name) | |
| # Small PutObject is a separate SigV4 path from large multipart cp. | |
| # The default uses aws-chunked + unsigned checksum trailers in | |
| # modern CLI v2; when_required sends a regular hashed payload. | |
| # Test both against the same sandbox endpoint and credential. | |
| single = folder / "single.txt" | |
| single.write_bytes(b"offline AWS CLI single-object signature test") | |
| for mode, overrides in ( | |
| ("streaming", {}), | |
| ("hashed", {"AWS_REQUEST_CHECKSUM_CALCULATION": "when_required"}), | |
| ): | |
| key = f"models/single-{mode}.txt" | |
| aws("s3api", "put-object", "--bucket", name, "--key", key, "--body", str(single), overrides=overrides) | |
| head = json.loads(aws("s3api", "head-object", "--bucket", name, "--key", key)) | |
| assert head["ContentLength"] == single.stat().st_size | |
| dest_single = folder / f"single-{mode}-download.txt" | |
| aws("s3", "cp", f"s3://{name}/{key}", str(dest_single), "--no-progress", overrides=overrides) | |
| assert dest_single.read_bytes() == single.read_bytes() | |
| signed_url = aws("s3", "presign", f"s3://{name}/{key}", "--expires-in", "60").strip() | |
| download = httpx.get(signed_url, verify=str(WORK / "localhost.pem"), trust_env=False) | |
| if download.status_code != 200: | |
| raise RuntimeError(f"Presigned GET failed: HTTP {download.status_code}") | |
| assert download.content == single.read_bytes() | |
| aws("s3", "rm", f"s3://{name}/{key}", overrides=overrides) | |
| # High-level small cp uses PutObject too; do not cover it only | |
| # with s3api or the separate large multipart transfer below. | |
| aws("s3", "cp", str(single), f"s3://{name}/{key}", "--no-progress", overrides=overrides) | |
| aws("s3", "rm", f"s3://{name}/{key}", overrides=overrides) | |
| aws("s3", "cp", str(source), f"s3://{name}/models/model.bin", "--no-progress") | |
| listing = aws("s3", "ls", f"s3://{name}/models/") | |
| assert "model.bin" in listing | |
| dest = folder / "download.bin" | |
| aws("s3", "cp", f"s3://{name}/models/model.bin", str(dest), "--no-progress") | |
| assert dest.read_bytes() == source.read_bytes() | |
| aws( | |
| "s3api", | |
| "put-object-tagging", | |
| "--bucket", | |
| name, | |
| "--key", | |
| "models/model.bin", | |
| "--tagging", | |
| json.dumps({"TagSet": [{"Key": "source", "Value": "multipart"}]}), | |
| ) | |
| aws("s3", "cp", f"s3://{name}/models/model.bin", f"s3://{name}/models/copied.bin", "--no-progress") | |
| copied_tags = json.loads(aws("s3api", "get-object-tagging", "--bucket", name, "--key", "models/copied.bin")) | |
| assert copied_tags["TagSet"] == [{"Key": "source", "Value": "multipart"}] | |
| copied = folder / "copied-download.bin" | |
| aws("s3", "cp", f"s3://{name}/models/copied.bin", str(copied), "--no-progress") | |
| assert copied.read_bytes() == source.read_bytes() | |
| aws("s3", "rm", f"s3://{name}/models/copied.bin") | |
| aws("s3", "rm", f"s3://{name}/models/model.bin") | |
| aws("s3", "cp", str(single), f"s3://{name}/copy-source.txt", "--no-progress") | |
| aws("s3", "cp", f"s3://{name}/copy-source.txt", f"s3://{name}/copy-target.txt", "--no-progress") | |
| aws("s3", "mv", f"s3://{name}/copy-target.txt", f"s3://{name}/moved.txt", "--no-progress") | |
| synced = folder / "synced" | |
| aws("s3", "sync", f"s3://{name}/", str(synced), "--no-progress") | |
| assert (synced / "moved.txt").read_bytes() == single.read_bytes() | |
| assert (synced / "copy-source.txt").read_bytes() == single.read_bytes() | |
| aws("s3", "rm", f"s3://{name}/", "--recursive") | |
| aws("s3api", "delete-bucket", "--bucket", name) | |
| print( | |
| "PASS AWS CLI v2: default/trailer and hashed single PutObject, multipart cp, server copy/move, sync, ls, download, recursive rm, delete" | |
| ) | |
| finally: | |
| native.delete("/v1/buckets/s3-credentials/" + credential["id"]).raise_for_status() | |
| native.close() | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 6.89 kB
- Xet hash:
- d7542c5786b4f7ae8e7aff52628c69c8ca55f1399ceaddbd67ba64ece91c4b7a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.