[Python] Patch Method Call Inside Instance method

Using pytest with Mock or pytest-mock, how would I prevent a instance method of a local variable inside an instance method from being called? Consider the following code:

from pydantic import SecretStr, BaseModel, Field

from google.cloud.storage_transfer_v1 import (
StorageTransferServiceClient,
TransferJob,
TransferSpec,
TransferManifest,
AwsS3Data,
AwsAccessKey,
GcsData,
RunTransferJobRequest,
CreateTransferJobRequest
)

from python_tools.consts import timestr
from python_tools.logging import logger

import time

class AwsStsGcp(BaseModel):
destination_bucket: str = Field(description='Destination GCS Bucket to transfer to.')
src_bucket: str = Field(description='AWS Source Bucket to transfer from.')
manifest_path: str = Field(description='Optional Path to manifest file.', default=None)
aws_access_key: SecretStr = Field(description="AWS Access Key to use for transfer.")
aws_secret_key: SecretStr = Field(description="AWS Secret Key to use for transfer.")
tranfer_name: str = Field(description="Name to use for Transfer Job. Timestamp will be appended.")
project_id: str = Field(description="GCP Project ID to use for Transfer Job.")

def create_aws_transfer_job(self) -> None:
"""create_aws_transfer_job: Transfer AWS S3 Data to GCS, using Storage Transfer Service.
"""
s3_config = None

transfer_manifest = None

client = StorageTransferServiceClient()
s3_config = AwsS3Data(
bucket_name=self.src_bucket,
aws_access_key=AwsAccessKey(
access_key_id=self.aws_access_key.get_secret_value(), secret_access_key=self.aws_secret_key.get_secret_value()
)
)

gcs_dest = GcsData(bucket_name=self.destination_bucket)

if self.manifest_path is not None:
transfer_manifest = TransferManifest(location=self.manifest_path)

sts_spec = TransferSpec(gcs_data_sink=gcs_dest, aws_s3_data_source=s3_config, transfer_manifest=transfer_manifest)
timestamp = time.strftime(timestr)
name = f"transferJobs/{self.tranfer_name}-{timestamp}"
description = "Automated STS Job created from Python Tools."
sts_job = TransferJob(
project_id=self.project_id,
name=name,
description=description,
transfer_spec=sts_spec,
status=TransferJob.Status.ENABLED,
)
job_request = CreateTransferJobRequest(transfer_job=sts_job)

logger.info(f"Starting Transfer Job for Job ID: {name}")
transfer_request = RunTransferJobRequest(project_id=self.project_id, job_name=name)

client.create_transfer_job(request=job_request)
client.run_transfer_job(request=transfer_request)


Test

import pytest
from unittest.mock import Mock
from python_tools.gcp.models.aws_sts_gcp import AwsStsGcp

def test_create_aws_transfer_job(mocker):
destination_bucket = "testbucket"
src_bucket = "testbucket"
tranfer_name = "test_transfer"
project_id = "test"
manifest_path = None
aws_access_key = "abc"
aws_secret_key = "123"
test_transfer = AwsStsGcp(aws_secret_key=aws_secret_key, aws_access_key=aws_access_key, project_id=project_id, tranfer_name=tranfer_name, destination_bucket=destination_bucket, src_bucket=src_bucket)

mocker.patch("create_aws_transfer_job.client.create_transfer_job")
mocker.patch("create_aws_transfer_job.client.run_transfer_job")
test_transfer.create_aws_transfer_job()


Is there a way to do something like this?

mocker.patch("create_aws_transfer_job.client.create_transfer_job")
mocker.patch("create_aws_transfer_job.client.run_transfer_job")

Continue reading...