test_storage.py

importtempfile
importtime

importpytest

fromskyimportexceptions
fromsky.dataimportstorageasstorage_lib


classTestStorageSpecLocalSource:
"""Tests for local sources"""

deftest_nonexist_local_source(self):
storage_obj=storage_lib.Storage(
name='test', source=f'/tmp/test-{int(time.time())}')
withpytest.raises(exceptions.StorageSourceError) ase:
storage_obj.construct()

assert'Local source path does not exist'instr(e)

deftest_source_trailing_slashes(self):
storage_obj=storage_lib.Storage(name='test', source='/bin/')
withpytest.raises(exceptions.StorageSourceError) ase:
storage_obj.construct()
assert'Storage source paths cannot end with a slash'instr(e)

deftest_source_single_file(self):
withtempfile.NamedTemporaryFile() asf:
storage_obj=storage_lib.Storage(name='test', source=f.name)
withpytest.raises(exceptions.StorageSourceError) ase:
storage_obj.construct()
assert'Storage source path cannot be a file'instr(e)

deftest_source_multifile_conflict(self):
storage_obj=storage_lib.Storage(
name='test', source=['/myfile.txt', '/a/myfile.txt'])
withpytest.raises(exceptions.StorageSourceError) ase:
storage_obj.construct()
assert'Cannot have multiple files or directories'instr(e)


classTestStorageSpecValidation:
"""Storage specification validation tests"""

# These tests do not create any buckets and can be run offline
deftest_source_and_name(self):
"""Tests when both name and source are specified"""
storage_obj=storage_lib.Storage(name='test', source='/bin')
storage_obj.construct()

# When source is bucket URL and name is specified - invalid spec
storage_obj=storage_lib.Storage(name='test',
source='s3://tcga-2-open')
withpytest.raises(exceptions.StorageSpecError) ase:
storage_obj.construct()

assert'Storage name should not be specified if the source is a ' \
'remote URI.'instr(e)

deftest_source_and_noname(self):
"""Tests when only source is specified"""
# When source is local, name must be specified
storage_obj=storage_lib.Storage(source='/bin')
withpytest.raises(exceptions.StorageNameError) ase:
storage_obj.construct()

assert'Storage name must be specified if the source is local'instr(e)

# When source is bucket URL and name is not specified - valid spec
# Cannot run this test because it requires AWS credentials to initialize
# bucket.
# storage_lib.Storage(source='s3://tcga-2-open')

deftest_name_and_nosource(self):
"""Tests when only name is specified"""
# When mode is COPY and the storage object doesn't exist - error out
storage_obj=storage_lib.Storage(name='sky-test-bucket',
mode=storage_lib.StorageMode.COPY)
withpytest.raises(exceptions.StorageSourceError) ase:
storage_obj.construct()

assert'source must be specified when using COPY mode'instr(e)

# When mode is MOUNT - valid spec (e.g., use for scratch space)
storage_obj=storage_lib.Storage(name='sky-test-bucket',
mode=storage_lib.StorageMode.MOUNT)
storage_obj.construct()

deftest_noname_and_nosource(self):
"""Tests when neither name nor source is specified"""
# Storage cannot be specified without name or source - invalid spec
storage_obj=storage_lib.Storage()
withpytest.raises(exceptions.StorageSpecError) ase:
storage_obj.construct()

assert'Storage source or storage name must be specified'instr(e)

deftest_uri_in_name(self):
"""Tests when name is a URI.

 Other tests for invalid names require store-specific test cases, and
 are in test_smoke.py::TestStorageWithCredentials"""
invalid_names= [
's3://mybucket',
'gs://mybucket',
'r2://mybucket',
 ]

fornininvalid_names:
storage_obj=storage_lib.Storage(name=n)
withpytest.raises(exceptions.StorageNameError) ase:
storage_obj.construct()

assert'Prefix detected'instr(e)