"""Fake subprocess.CompletedProcess class used for mocking return

def __init__(self, stdout: str = '',

stderr: str = '',

returncode: int = 0,

subprocess_run_called: bool = True,

storage: dict[str, str] | None = None,

key: str | None = None):

"""Class constructor

if subprocess_run_called:

self.stdout: bytes | None = str.encode(stdout)

self.stderr: bytes | None = str.encode(stderr)

else:

# when CompletedProcess is created by subprocess.Popen,

# CompletedProcess.stdout and CompletedProcess.stderr are streams

self.stdout = io.StringIO(stdout)

self.stderr = io.StringIO(stderr)

self.returncode: int = returncode

self.stdin = MagicMock()

self.storage = storage

self.key = key

def communicate(self, arg):

"""Simulate writing subprocess stdin to a cloud storage object.

if self.storage and self.key:

if NOT_WRITABLE_BUCKET in self.key:

return (1, b'Mocked error: cannot write to bucket')

self.storage[self.key] = arg.decode()

else:

return (1, b'Mocked error: no storage or key')

"""Fixtire function that replaces util.safe_exec with mocked_safe_exec"""

mock = GKEMock()

mock.cloud.conf['project'] = GCP_PROJECT

mock.cloud.storage['gs://test-bucket'] = 0

mock.cloud.storage['gs://test-bucket/test-query.fa'] = '>query\nACTGGAGATGAC'

mock.cloud.storage['gs://test-results'] = ''

mock.cloud.storage[f'gs://{NOT_WRITABLE_BUCKET}'] = ''

mock.cloud.storage['s3://test-bucket/test-query.fa'] = '>query\nACTGGAGATGAC'

mock.cloud.storage['s3://test-bucket'] = 0

mock.cloud.storage['s3://test-results'] = ''

mock.cloud.storage[f's3://{NOT_WRITABLE_BUCKET}'] = ''

# Mocked NCBI database metadata

mock.cloud.storage['gs://blast-db/latest-dir'] = '000'

mock.cloud.storage[f'gs://blast-db/000/{DB_METADATA_PROT_FILE_NAME}'] = DB_METADATA_PROT

mock.cloud.storage[f'gs://blast-db/000/{DB_METADATA_NUCL_FILE_NAME}'] = DB_METADATA_NUCL

mock.cloud.storage['s3://ncbi-blast-databases/latest-dir'] = '000'

mock.cloud.storage[f's3://ncbi-blast-databases/000/{DB_METADATA_PROT_FILE_NAME}'] = DB_METADATA_PROT

mock.cloud.storage[f's3://ncbi-blast-databases/000/{DB_METADATA_NUCL_FILE_NAME}'] = DB_METADATA_NUCL

# User database metadata

mock.cloud.storage[f'gs://test-bucket/{DB_METADATA_PROT_FILE_NAME}'] = DB_METADATA_PROT

mock.cloud.storage[f'gs://test-bucket/{DB_METADATA_NUCL_FILE_NAME}'] = DB_METADATA_NUCL

mock.cloud.storage['gs://test-bucket/testdb.pal'] = 'A fake user database'

mock.cloud.storage[f's3://test-bucket/{DB_METADATA_PROT_FILE_NAME}'] = DB_METADATA_PROT

mock.cloud.storage[f's3://test-bucket/{DB_METADATA_NUCL_FILE_NAME}'] = DB_METADATA_NUCL

mock.cloud.storage['s3://test-bucket/testdb.pal'] = 'A fake user database'

# we need gcp.safe_exec instead of util.safe exec here, because

# safe_exec is imported in gcp.py with 'from util import safe_exec'

# and safe_exec in gcp is seen as local, python is funny this way

mocker.patch('elastic_blast.gcp.safe_exec', side_effect=mock.mocked_safe_exec)

mocker.patch('elastic_blast.kubernetes.safe_exec', side_effect=mock.mocked_safe_exec)

mocker.patch('elastic_blast.util.safe_exec', side_effect=mock.mocked_safe_exec)

mocker.patch('elastic_blast.filehelper.safe_exec', side_effect=mock.mocked_safe_exec)

mocker.patch('elastic_blast.elb_config.safe_exec', side_effect=mock.mocked_safe_exec)

mocker.patch('elastic_blast.gcp_traits.safe_exec', side_effect=mock.mocked_safe_exec)

mocker.patch('elastic_blast.tuner.aws_get_machine_type', new=MagicMock(return_value='test-machine-type'))

mocker.patch('subprocess.Popen', side_effect=mock.mocked_popen)

mocker.patch('boto3.resource', side_effect=mock.mocked_resource)

mocker.patch('boto3.client', side_effect=mock.mocked_client)

mocker.patch('botocore.exceptions.ClientError.__init__', new=MagicMock(return_value=None))

mocker.patch.dict(os.environ, {'ELB_PAUSE_AFTER_INIT_PV': '1'})

mocker.patch('shutil.which', side_effect=MagicMock(return_value='.'))

yield mock

"""Generate config for mocked gcloud and kubeclt"""

cfg = ElasticBlastConfig(gcp_project = GCP_PROJECT,

gcp_region = GCP_REGION,

gcp_zone = GCP_ZONE,

program = 'blastn',

db = 'testdb',

queries = 'test-queries.fa',

results = 'gs://elasticblast-blastadm',

task = ElbCommand.SUBMIT)

cfg.cluster.name = GKE_CLUSTERS[0]

"""Class to simulate created cloud resources"""

# dictionary of cloud storage objects, where key is object key and value is

# object content, any object is readable and writable

storage: dict[str, str] = field(default_factory=dict)

# gcloud config

"""Substitute for util.safe_exec function that calls command line gcloud

if isinstance(cmd, str):

cmd = cmd.split()

if not isinstance(cmd, list):

raise ValueError('Argument to mocked_safe_exec must be a list or string')

# check if gcloud is installed

if ' '.join(cmd) == 'gcloud --version':

return MockedCompletedProcess()

# get GCP project

if ' '.join(cmd) == 'gcloud config get-value project':

return MockedCompletedProcess(GCP_PROJECT)

# get GCP account name

elif ' '.join(cmd) == 'gcloud config get-value account':

return MockedCompletedProcess('the-gcp-account-name')

# set GCP project

elif ' '.join(cmd[:-1]) == 'gcloud config set project':

return MockedCompletedProcess()

# get a list of presistent disks

elif ' '.join(cmd).startswith('gcloud compute disks list --format json'):

result = []

for d in GCP_DISKS:

result.append({'name': d})

return MockedCompletedProcess(json.dumps(result))

# delete a persistent disk

elif ' '.join(cmd).startswith('gcloud compute disks delete'):

return MockedCompletedProcess()

# get a list of volume snapshots

elif ' '.join(cmd).startswith('gcloud compute snapshots list --format json'):

result = [{'name': 'snapshot-12345'}]

return MockedCompletedProcess(json.dumps(result))

# delete a volume snapshot

elif ' '.join(cmd).startswith('gcloud compute snapshots delete'):

return MockedCompletedProcess()

# GKE cluster status

elif ' '.join(cmd).startswith('gcloud container clusters list --format=value(status) --filter name'):

return MockedCompletedProcess('RUNNING\n')

# list GKE clusters

elif ' '.join(cmd).startswith('gcloud container clusters list --format json'):

result = []

for i in GKE_CLUSTERS:

result.append({'name': i})

return MockedCompletedProcess(json.dumps(result))

# get GKE cluster credentials

elif ' '.join(cmd).startswith('gcloud container clusters get-credentials'):

return MockedCompletedProcess()

# delete cluster

elif ' '.join(cmd).startswith('gcloud container clusters delete'):

return MockedCompletedProcess()

# Get GCP regions

elif ' '.join(cmd).startswith('gcloud compute regions list'):

return MockedCompletedProcess('[{"name":"us-east4"},{"name":"us-east1"},{"name":"test-gcp-region"},{"name":"mocked-gcp-region"},{"name": "test-region"}]')

# check GCP APIs

elif ' '.join(cmd).startswith('gcloud services '):

return MockedCompletedProcess()

# get persistent disks

elif cmd[0] == 'kubectl' and 'get pv -o json' in ' '.join(cmd):

result = {'items': []} # type: ignore

for i in GCP_DISKS:

result['items'].append({'spec': {'csi': {'volumeHandle': f'/test-project/test-region/{i}'}}}) # type: ignore

return MockedCompletedProcess(json.dumps(result))

# get volume snapshots

elif cmd[0] == 'kubectl' and 'get volumesnapshot' in ' '.join(cmd):

result = {'items': [{'metadata': {'uid': '12345'}}]}

return MockedCompletedProcess(json.dumps(result))

# get kubernetes jobs

elif cmd[0] == 'kubectl' and 'get jobs -o json' in ' '.join(cmd):

result = {'items': []}

for i in K8S_JOBS:

result['items'].append({'metadata': {'name': i}})

return MockedCompletedProcess(json.dumps(result))

elif cmd[0] == 'kubectl' and 'get pv,pvc -o=NAME' in ' '.join(cmd):

return MockedCompletedProcess('persistentvolume/pvc-3aafea07-4d87-4349-bcfa-fce4cf8c0197')

elif cmd[0] == 'kubectl' and 'get pv -o custom-columns=CLAIM:' in ' '.join(cmd):

return MockedCompletedProcess(stdout='CLAIM PDNAME\nblast-dbs-pvc gke-some-synthetic-name')

elif cmd[0] == 'kubectl' and 'describe' in ' '.join(cmd):

return MockedCompletedProcess('')

elif cmd[0] == 'kubectl' and 'patch' in ' '.join(cmd):

return MockedCompletedProcess('')

# get kubernetes job status with one pod deleted due to failure

elif cmd[0] == 'kubectl' and 'get pods -o custom-columns=STATUS' in ' '.join(cmd):

return MockedCompletedProcess('\n'.join(['STATUS'] + ['Running' for i in K8S_JOB_STATUS if i == 'Running']))

elif cmd[0] == 'kubectl' and 'get jobs' in ' '.join(cmd):

switcher = {'Failed': 'Failed',

'Succeeded': 'Complete',

'Running': '<none>',

'Pending': '<none>'}

return MockedCompletedProcess('\n'.join(['STATUS'] + [switcher[i] for i in K8S_JOB_STATUS]))

# delete all jobs

elif cmd[0] == 'kubectl' and 'delete jobs' in ' '.join(cmd):

return MockedCompletedProcess('\n'.join(['deleted ' + i for i in K8S_JOBS]) + '\n')

# delete all pvcs

elif cmd[0] == 'kubectl' and 'delete pvc --all' in ' '.join(cmd):

return MockedCompletedProcess('\n'.join(['deleted ' + i for i in GKE_PVS]) + '\n')

# delete all pvs

elif cmd[0] == 'kubectl' and 'delete pv --all' in ' '.join(cmd):

return MockedCompletedProcess('\n')

# delete all volume snapshots

elif cmd[0] == 'kubectl' and 'delete volumesnapshots --all' in ' '.join(cmd):

return MockedCompletedProcess('\n')

# check if kubernetes client is installed or cluster is alive

elif ' '.join(cmd).startswith('kubectl') and 'version' in ' '.join(cmd):

return MockedCompletedProcess('{ "clientVersion": { "major": "1", "minor": "27", "gitVersion": "v1.27.4", "gitCommit": "286cfa5f978c4a89c776347c82fa09a232eef144", "gitTreeState": "clean", "buildDate": "2024-03-06T00:56:29Z", "goVersion": "go1.20.12 X:strictfipsruntime", "compiler": "gc", "platform": "linux/amd64" }, "kustomizeVersion": "v5.0.1" }')

# delete a kubernetes resopurce by file

elif cmd[0] == 'kubectl' and 'delete -f' in ' '.join(cmd):

return MockedCompletedProcess()

# list BLAST databases available in the cloud

elif ' '.join(cmd).startswith('update_blastdb.pl') and '--showall' in cmd:

return MockedCompletedProcess(f'{BLASTDB}\tTitle of {BLASTDB}\t0.1\t2020-01-01')

# Check the resource quota in GKE

elif ' '.join(cmd).startswith('kubectl get resourcequota gke-resource-quotas'):

return MockedCompletedProcess('"10k"')

# check if gsutil is installed

elif ' '.join(cmd) == 'gsutil --version':

return MockedCompletedProcess()

# Check whether a file exists in GCS

elif ' '.join(cmd).startswith('gsutil') and 'stat' in cmd:

if cloud_state:

# handle a wildcard '*'

if cmd[-1].rstrip().endswith('*'):

for key in cloud_state.storage:

if key.startswith(cmd[-1][:-1]):

return MockedCompletedProcess()

raise SafeExecError(returncode=1, message=f'File {cmd[-1]} was not found')

# handle an exact name

if cmd[-1] in cloud_state.storage:

return MockedCompletedProcess()

else:

raise SafeExecError(returncode=1, message=f'File {cmd[-1]} was not found')

else:

return MockedCompletedProcess(stdout='',stderr='',returncode=0)

# Check whether a file exists in GCS

elif ' '.join(cmd).startswith('gsutil') and 'cat' in cmd:

cmd = ' '.join(cmd)

# simulate reading NCBI database manifest

if cmd.endswith('latest-dir'):

return MockedCompletedProcess(stdout='xxxx')

elif cmd.endswith('blastdb-metadata-1-1.json'):

manifest = {'nr': {'size': 25}, 'nt': {'size': 25}, 'pdbnt': {'size': 25}, 'testdb': {'size': 25}}

return MockedCompletedProcess(stdout=json.dumps(manifest))

else:

return MockedCompletedProcess(stdout='',stderr='',returncode=0)

# copy files to GCS

elif ' '.join(cmd).startswith('gsutil') and 'cp' in cmd:

return MockedCompletedProcess()

# Get file length on GCS

elif ' '.join(cmd).startswith('gsutil') and 'ls' in cmd:

return MockedCompletedProcess(stdout=str(ELB_DFLT_FSIZE_FOR_TESTING))

# remove a file from GCS

elif ' '.join(cmd).startswith('gsutil') and 'rm' in cmd:

return MockedCompletedProcess(stdout='',stderr='',returncode=0)

elif ' '.join(cmd).startswith('kubectl config current-context'):

return MockedCompletedProcess(stdout='dummy-context',stderr='',returncode=0)

elif cmd[0] == 'kubectl' and 'apply -f' in ' '.join(cmd):

return MockedCompletedProcess(stdout='',stderr='',returncode=0)

elif ' '.join(cmd).startswith('gcloud compute regions describe'):

return MockedCompletedProcess(stdout='{"quotas":[{"limit": 81920.0,"metric": "SSD_TOTAL_GB","usage": 2666.0}]}',stderr='',returncode=0)

# raise ValueError for unrecognized command line

else:

"""Utility class for mocking util.safe_exec that with different

allowed_options = ['no-cluster',

'kubectl-error',

'bad-results-bucket']

def __init__(self):

"""Class constructor"""

self.options = list()

self.disk_delete_called = False

self.cloud = CloudResources()

def set_options(self, options: list[str]) -> None:

"""Set optional mocked GKE behavior.

self.options = options

for opt in options:

if opt not in GKEMock.allowed_options:

raise ValueError(f'Unsupported GKEMock option: {opt}')

def mocked_safe_exec(self, cmd, env = None):

"""Mocked util.safe_exec function"""

if isinstance(cmd, list):

cmd = ' '.join(cmd)

# simulate manipulation of a non-existent GKE cluster

if 'no-cluster' in self.options:

if cmd.startswith('gcloud container clusters get-credentials') or \

cmd.startswith('gcloud container clusters delete'):

raise SafeExecError(returncode=1,

message='Mocked error: cluster does not exist')

if cmd.startswith('gcloud container clusters delete'):

raise UserReportError(returncode=CLUSTER_ERROR,

message='Mocked error: cluster does not exist')

if cmd.startswith('gcloud container clusters list'):

return MockedCompletedProcess('')

# any kubectl call fails

if 'kubectl-error' in self.options:

if cmd.startswith('kubectl') and 'config current-context' not in cmd:

raise SafeExecError(returncode=1,

message=f'Mocked kubectl error in "{cmd}"')

# report no disks after disk deletion was called

if cmd.startswith('gcloud compute disks list') and self.disk_delete_called:

return MockedCompletedProcess(json.dumps([]))

if cmd.startswith('gcloud compute disks delete'):

self.disk_delete_called = True

if cmd.startswith('gcloud config get-value project'):

print(self.cloud.conf)

if 'project' in self.cloud.conf and self.cloud.conf['project']:

return MockedCompletedProcess(self.cloud.conf['project'])

return MockedCompletedProcess('(unset)')

return mocked_safe_exec(cmd, env=env, cloud_state=self.cloud)

def mocked_popen(self, cmd, stderr, stdin=None, stdout=None, universal_newlines=True):

"""Mocked subprocess.Popen function, used to mock calls to gsutil used

# open_for_read

if ' '.join(cmd).startswith('gsutil') and 'cat' in cmd:

if cmd[-1] in self.cloud.storage:

return MockedCompletedProcess(stdout=self.cloud.storage[cmd[-1]], stderr='', subprocess_run_called=False)

else:

return MockedCompletedProcess(returncode=1, stdout='', stderr=f'Object "{cmd[-1]}" does not exist', subprocess_run_called=False)

# test dir for write

elif ' '.join(cmd).startswith('gsutil') and 'cp' in cmd and '-' in cmd:

if '/'.join(cmd[-1].split('/')[:-1]) in self.cloud.storage:

if NOT_WRITABLE_BUCKET in cmd[-1]:

return MockedCompletedProcess(returncode=1, stderr=f'Mocked error: cannot write to bucker {cmd[-1]}')

return MockedCompletedProcess(storage=self.cloud.storage, key=cmd[-1], subprocess_run_called=False)

else:

return MockedCompletedProcess(returncode=1, subprocess_run_called=False)

# raise ValueError for unrecognized command line

else:

raise ValueError(f'Unrecognized gcloud or kubectl command line: {cmd}')

def mocked_resource(self, resource, config=None):

"""Mocked boto3.resource function"""

if resource == 's3':

return MockedS3Resource(self.cloud.storage)

elif resource == 'cloudformation':

return MockedCloudformationResource()

else:

raise NotImplementedError(f'boto3 mock for {resource} resource is not implemented')

def mocked_client(self, client, config=None):

"""Mocked boto3.resource function"""

if client == 's3':

return MockedS3Client(self.cloud.storage)

elif client == 'sts':

return MockedStsClient()

elif client == 'ec2':

return MockedEC2Client()

else:

"""Credentials for mocked AWS services. This fixture ensures that we are

# Setup

# save AWS-related variables before modifying environment

saved_vars = {key: os.environ[key] for key in os.environ if key.startswith('AWS')}

os.environ['AWS_ACCESS_KEY_ID'] = 'testing'

os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'

os.environ['AWS_SESSION_TOKEN'] = 'testing'

os.environ['AWS_ACCT'] = 'testing'

os.environ['AWS_DEFAULT_REGION'] = ELB_DFLT_AWS_REGION

yield

# Cleanup

# bring back pre-test environment

for i in saved_vars:

"""Mocked boto3 S3 object"""

def __init__(self, bucket, key):

self.obj = f's3://{bucket}/{key}'

self.storage = {}

self.content_length = 123456789

def load(self):

"""Raise ClientError if the object is not in storage, otherwise do nothing"""

if self.obj not in self.storage:

raise ClientError(None, None)

def upload_fileobj(self, stream, Config = None):

"""Upload a file object to the cloud bucket"""

self.storage[self.obj] = stream.read()

def download_fileobj(self, stream):

"""Download a file object from the cloud bucket"""

""" Mocked EC2 client """

def describe_regions(self):

retval = { 'Regions': [ ] }

for r in AWS_REGIONS:

retval['Regions'].append({'RegionName': r})

"""Mocked boto3 S3 resource object"""

def __init__(self, cloud_storage):

"""Initialize object"""

self.storage = cloud_storage

def Object(self, bucket, key):

"""Mocked s3.Object function that creates S3 objects"""

obj = MockedS3Object(bucket, key)

obj.storage = self.storage

"""A string stream class needed for mocked downloads from S3, used by

def __init__(self, data):

"""Initialize an object"""

self.data = data

self.cl = False

def close(self):

"""Close stream"""

pass

def read(self, pos):

"""Read from the stream"""

if not self.cl:

self.cl = True

return self.data.encode()

else:

"""Mocked boto3 S3 client object"""

def __init__(self, cloud_storage):

"""Initialize object"""

self.storage = cloud_storage

def get_object(self, Bucket, Key):

"""Get an S3 object"""

key = f's3://{Bucket}/{Key}'

if key not in self.storage:

raise

"""Mocked boto3 STS client object"""

def get_caller_identity(self):

"""Get IAM user or role name"""

return {'UserId': 'test-user',

'Account': 'test-account',

"""Mocked boto3 cloudformation stack object"""

def __init__(self, name):

self.name = name

def __getattr__(self, name):

"""Always raise an exception to indicate that there is no stack named

"""Mocked boto3 cloudformation resource"""

def Stack(self, name):

"""Mocked create stack object"""