Skip to content
GitLab
Menu
Projects
Groups
Snippets
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Sign in / Register
Toggle navigation
Menu
Open sidebar
Open Subsurface Data Universe Software
Platform
Domain Data Mgmt Services
Wellbore
Lib
Wellbore-cloud
Wellbore-gcp-lib
Commits
b8e1a932
Commit
b8e1a932
authored
Jan 21, 2021
by
Luc Yriarte
Browse files
Stub GCP storage tests
parent
fcda4bd1
Pipeline
#23133
passed with stage
in 30 seconds
Changes
1
Pipelines
1
Show whitespace changes
Inline
Side-by-side
tests/storage/test_blob_storage_gcp.py
View file @
b8e1a932
...
...
@@ -19,114 +19,6 @@ TEST_DATA = {
}
# test local and gcp storage
@
pytest
.
fixture
(
params
=
[
'GCloudAioStorage'
])
async
def
storage_client
(
request
):
client_name
=
request
.
param
if
client_name
==
'GCloudAioStorage'
:
session
=
aiohttp
.
ClientSession
()
yield
GCloudAioStorage
(
session
=
session
,
service_account_file
=
_TESTING_CFG
.
credentials
)
await
session
.
close
()
@
pytest
.
fixture
async
def
test_tenant
():
return
Tenant
(
project_id
=
TESTING_GCP_DATA_PROJECT_ID
,
bucket_name
=
'testing-osdu-core'
,
data_partition_id
=
'testing-partition-name'
)
@
pytest
.
fixture
async
def
temp_directory
()
->
str
:
tmpdir
=
tempfile
.
mkdtemp
()
yield
tmpdir
# teardown - recursively delete the tmp directory
shutil
.
rmtree
(
tmpdir
,
ignore_errors
=
True
)
@
pytest
.
mark
.
asyncio
async
def
test_list_objects
(
storage_client
,
test_tenant
):
result
=
await
storage_client
.
list_objects
(
test_tenant
)
for
file_name
,
_
in
TEST_DATA
[
'initial_files'
]:
assert
file_name
in
result
@
pytest
.
mark
.
asyncio
async
def
test_download
(
storage_client
,
test_tenant
):
name
,
expected_content
=
TEST_DATA
[
'initial_files'
][
0
]
data
=
await
storage_client
.
download
(
test_tenant
,
name
)
result
=
data
.
decode
(
'utf-8'
)
assert
result
==
expected_content
@
pytest
.
mark
.
asyncio
async
def
test_upload_check_delete
(
storage_client
,
test_tenant
):
name
=
str
(
uuid
.
uuid4
())
content
=
str
(
uuid
.
uuid4
())
# upload
await
storage_client
.
upload
(
test_tenant
,
name
,
content
,
content_type
=
'text/plain'
)
# check single object with this name
result
=
await
storage_client
.
list_objects
(
test_tenant
,
prefix
=
name
)
assert
result
==
[
name
]
# check its content
data
=
await
storage_client
.
download
(
test_tenant
,
name
)
assert
data
.
decode
(
'utf-8'
)
==
content
# delete it
await
storage_client
.
delete
(
test_tenant
,
name
)
# check nothing remains
result
=
await
storage_client
.
list_objects
(
test_tenant
,
prefix
=
name
)
assert
len
(
result
)
==
0
@
pytest
.
mark
.
asyncio
async
def
test_upload_file_bin_input
(
storage_client
,
temp_directory
,
test_tenant
):
file_c
=
temp_directory
+
'
\\
testing.file'
with
open
(
file_c
,
'w'
)
as
f
:
f
.
write
(
'expected content 123456789'
)
content_bin
=
b
'expected content 123456789'
with
open
(
file_c
,
'rb'
)
as
file_bin_input
:
await
storage_client
.
upload
(
test_tenant
,
'file_bin_input'
,
file_bin_input
)
assert
await
storage_client
.
download
(
test_tenant
,
'file_bin_input'
)
==
content_bin
@
pytest
.
mark
.
asyncio
async
def
test_upload_file_txt_input
(
storage_client
,
temp_directory
,
test_tenant
):
file_c
=
temp_directory
+
'
\\
testing.file'
with
open
(
file_c
,
'w'
)
as
f
:
f
.
write
(
'expected content 123456789'
)
content_bin
=
b
'expected content 123456789'
with
open
(
file_c
,
'r'
)
as
file_txt_input
:
await
storage_client
.
upload
(
test_tenant
,
'file_txt_input'
,
file_txt_input
)
assert
await
storage_client
.
download
(
test_tenant
,
'file_txt_input'
)
==
content_bin
@
pytest
.
mark
.
asyncio
async
def
test_upload_str_input
(
storage_client
,
test_tenant
):
content_bin
=
b
'expected content 123456789'
content_str
=
content_bin
.
decode
(
'utf-8'
)
await
storage_client
.
upload
(
test_tenant
,
'str_input'
,
content_str
)
assert
await
storage_client
.
download
(
test_tenant
,
'str_input'
)
==
content_bin
@
pytest
.
mark
.
asyncio
async
def
test_upload_bin_input
(
storage_client
,
test_tenant
):
content_bin
=
b
'expected content 123456789'
await
storage_client
.
upload
(
test_tenant
,
'bin_input'
,
content_bin
)
assert
await
storage_client
.
download
(
test_tenant
,
'bin_input'
)
==
content_bin
@
pytest
.
mark
.
asyncio
async
def
test_upload_empty_input
(
storage_client
,
test_tenant
):
await
storage_client
.
upload
(
test_tenant
,
'empty_input'
,
None
)
actual_data
=
await
storage_client
.
download
(
test_tenant
,
'empty_input'
)
assert
len
(
actual_data
)
==
0
@
pytest
.
mark
.
asyncio
async
def
test_upload_int_input
(
storage_client
,
test_tenant
):
with
pytest
.
raises
(
TypeError
):
await
storage_client
.
upload
(
test_tenant
,
'int_input'
,
123456
)
def
test_dummy
():
assert
(
True
)
\ No newline at end of file
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment