nixos/nix
1
1
Fork 0
mirror of https://github.com/NixOS/nix.git synced 2025-11-09 03:56:01 +01:00
Code Activity

refactor(tests/nixos/s3-binary-cache-store): rename populate_with to populate_bucket

Browse source
This commit is contained in:
Bernardo Meurer Costa 2025-10-18 23:27:03 +00:00
parent c1a15d1a26
commit 9058d90ab2
No known key found for this signature in database
1 changed files with 9 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

18
tests/nixos/s3-binary-cache-store.nix
View file

@ -135,22 +135,22 @@ in
print(output) print(output)
raise Exception(f"{error_msg}: expected {expected}, got {actual}") raise Exception(f"{error_msg}: expected {expected}, got {actual}")
def setup_s3(populate_with=[]): def setup_s3(populate_bucket=[]):
""" """
Decorator that creates/destroys a unique bucket for each test. Decorator that creates/destroys a unique bucket for each test.
Optionally pre-populates bucket with specified packages. Optionally pre-populates bucket with specified packages.
Args: Args:
populate_with: List of packages to upload before test runs populate_bucket: List of packages to upload before test runs
""" """
def decorator(test_func): def decorator(test_func):
def wrapper(): def wrapper():
bucket = str(uuid.uuid4()) bucket = str(uuid.uuid4())
server.succeed(f"mc mb minio/{bucket}") server.succeed(f"mc mb minio/{bucket}")
try: try:
if populate_with: if populate_bucket:
store_url = make_s3_url(bucket) store_url = make_s3_url(bucket)
for pkg in populate_with: for pkg in populate_bucket:
server.succeed(f"{ENV_WITH_CREDS} nix copy --to '{store_url}' {pkg}") server.succeed(f"{ENV_WITH_CREDS} nix copy --to '{store_url}' {pkg}")
test_func(bucket) test_func(bucket)
finally: finally:
@ -182,7 +182,7 @@ in
print("✓ Credential provider created once and cached") print("✓ Credential provider created once and cached")
@setup_s3(populate_with=[PKGS['A']]) @setup_s3(populate_bucket=[PKGS['A']])
def test_fetchurl_basic(bucket): def test_fetchurl_basic(bucket):
"""Test builtins.fetchurl works with s3:// URLs""" """Test builtins.fetchurl works with s3:// URLs"""
print("\n=== Testing builtins.fetchurl ===") print("\n=== Testing builtins.fetchurl ===")
@ -218,7 +218,7 @@ in
print("✓ Error messages format URLs correctly") print("✓ Error messages format URLs correctly")
@setup_s3(populate_with=[PKGS['A']]) @setup_s3(populate_bucket=[PKGS['A']])
def test_fork_credential_preresolution(bucket): def test_fork_credential_preresolution(bucket):
"""Test credential pre-resolution in forked processes""" """Test credential pre-resolution in forked processes"""
print("\n=== Testing Fork Credential Pre-resolution ===") print("\n=== Testing Fork Credential Pre-resolution ===")
@ -298,7 +298,7 @@ in
print(" ✓ Child uses pre-resolved credentials (no new providers)") print(" ✓ Child uses pre-resolved credentials (no new providers)")
@setup_s3(populate_with=[PKGS['A'], PKGS['B'], PKGS['C']]) @setup_s3(populate_bucket=[PKGS['A'], PKGS['B'], PKGS['C']])
def test_store_operations(bucket): def test_store_operations(bucket):
"""Test nix store info and copy operations""" """Test nix store info and copy operations"""
print("\n=== Testing Store Operations ===") print("\n=== Testing Store Operations ===")
@ -337,7 +337,7 @@ in
print(" ✓ nix copy works") print(" ✓ nix copy works")
print(" ✓ Credentials cached on client") print(" ✓ Credentials cached on client")
@setup_s3(populate_with=[PKGS['A']]) @setup_s3(populate_bucket=[PKGS['A']])
def test_url_format_variations(bucket): def test_url_format_variations(bucket):
"""Test different S3 URL parameter combinations""" """Test different S3 URL parameter combinations"""
print("\n=== Testing URL Format Variations ===") print("\n=== Testing URL Format Variations ===")
@ -352,7 +352,7 @@ in
client.succeed(f"{ENV_WITH_CREDS} nix store info --store '{url2}' >&2") client.succeed(f"{ENV_WITH_CREDS} nix store info --store '{url2}' >&2")
print(" ✓ Parameter order: endpoint before region works") print(" ✓ Parameter order: endpoint before region works")
@setup_s3(populate_with=[PKGS['A']]) @setup_s3(populate_bucket=[PKGS['A']])
def test_concurrent_fetches(bucket): def test_concurrent_fetches(bucket):
"""Validate thread safety with concurrent S3 operations""" """Validate thread safety with concurrent S3 operations"""
print("\n=== Testing Concurrent Fetches ===") print("\n=== Testing Concurrent Fetches ===")