/
test_google_lifesciences.py
111 lines (92 loc) · 3.45 KB
/
test_google_lifesciences.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import os
import sys
import tempfile
from google.cloud import storage
sys.path.insert(0, os.path.dirname(__file__))
from common import *
def has_google_credentials():
return "GOOGLE_APPLICATION_CREDENTIALS" in os.environ
google_credentials = pytest.mark.skipif(
not has_google_credentials(),
reason="Skipping google lifesciences tests because "
"GOOGLE_APPLICATION_CREDENTIALS not found in the environment.",
)
def cleanup_google_storage(prefix, bucket_name="snakemake-testing"):
"""Given a storage prefix and a bucket, recursively delete files there
This is intended to run after testing to ensure that
the bucket is cleaned up.
Arguments:
prefix (str) : the "subfolder" or prefix for some files in the buckets
bucket_name (str) : the name of the bucket, default snakemake-testing
"""
client = storage.Client()
bucket = client.get_bucket(bucket_name)
blobs = bucket.list_blobs(prefix="source")
for blob in blobs:
blob.delete()
# Using API we get an exception about bucket deletion
shell("gsutil -m rm -r gs://{bucket.name}/* || true")
bucket.delete()
def create_google_storage(bucket_name="snakemake-testing"):
"""Given a bucket name, create the Google storage bucket,
intending to be used for testing and then cleaned up by
cleanup_google_storage
Arguments:
bucket_name (str) : the name of the bucket, default snakemake-testing
"""
client = storage.Client()
return client.create_bucket(bucket_name)
@google_credentials
def test_google_lifesciences():
bucket_name = "snakemake-testing-%s" % next(tempfile._get_candidate_names())
create_google_storage(bucket_name)
storage_prefix = "test_google_lifesciences"
workdir = dpath("test_google_lifesciences")
try:
run(
workdir,
use_conda=True,
default_remote_prefix="%s/%s" % (bucket_name, storage_prefix),
google_lifesciences=True,
google_lifesciences_cache=False,
preemption_default=None,
preemptible_rules=["pack=1"],
)
finally:
cleanup_google_storage(storage_prefix, bucket_name)
@pytest.mark.skip(
reason="Cannot test using touch with a remote prefix until the container image is deployed."
)
@google_credentials
def test_touch_remote_prefix():
bucket_name = "snakemake-testing-%s" % next(tempfile._get_candidate_names())
create_google_storage(bucket_name)
storage_prefix = "test_touch_remote_prefix"
workdir = dpath("test_touch_remote_prefix")
try:
run(
workdir,
use_conda=True,
default_remote_prefix="%s/%s" % (bucket_name, storage_prefix),
google_lifesciences=True,
google_lifesciences_cache=False,
)
finally:
cleanup_google_storage(storage_prefix, bucket_name)
@google_credentials
def test_cloud_checkpoints_issue574():
"""see Github issue #574"""
bucket_name = "snakemake-testing-%s" % next(tempfile._get_candidate_names())
create_google_storage(bucket_name)
storage_prefix = "test_cloud_checkpoints_issue574"
workdir = dpath("test_cloud_checkpoints_issue574")
try:
run(
workdir,
use_conda=True,
default_remote_prefix="%s/%s" % (bucket_name, storage_prefix),
google_lifesciences=True,
google_lifesciences_cache=False,
)
finally:
cleanup_google_storage(storage_prefix, bucket_name)