[Python] create unittest for a function that use API request to google cloud logging API

First I have to explain that I'm not a software engineer so all of this concept are new to me. I try to submit a PR to a repo to improve a feature, the feature works well but now but I need to implement a test which I'm not really used to. And this test is for a function that requires to use a google cloud library. I don't really understand how I should use a mock or a patch in my test to replace the result of the request to Cloud Logging API.

To help you understand my problem here is the function fetch_remote_logs that I created:

class DbtGcpCloudRunJobBaseOperator(AbstractDbtBase, CloudRunExecuteJobOperator): # type: ignore
"""
Executes a dbt core cli command in a Cloud Run Job instance with dbt installed in it.

"""

template_fields: Sequence[str] = tuple(
list(AbstractDbtBase.template_fields) + list(CloudRunExecuteJobOperator.template_fields)
)

intercept_flag = False

def __init__(
self,
# arguments required by CloudRunExecuteJobOperator
project_id: str,
region: str,
job_name: str,
#
profile_config: ProfileConfig | None = None,
command: list[str] | None = None,
environment_variables: dict[str, Any] | None = None,
**kwargs: Any,
) -> None:
self.profile_config = profile_config
self.command = command
self.environment_variables = environment_variables or DEFAULT_ENVIRONMENT_VARIABLES
super().__init__(project_id=project_id, region=region, job_name=job_name, **kwargs)
# In PR #1474, we refactored cosmos.operators.base.AbstractDbtBase to remove its inheritance from BaseOperator
# and eliminated the super().__init__() call. This change was made to resolve conflicts in parent class
# initializations while adding support for ExecutionMode.AIRFLOW_ASYNC. Operators under this mode inherit
# Airflow provider operators that enable deferrable SQL query execution. Since super().__init__() was removed
# from AbstractDbtBase and different parent classes require distinct initialization arguments, we explicitly
# initialize them (including the BaseOperator) here by segregating the required arguments for each parent class.
kwargs.update(
{
"project_id": project_id,
"region": region,
"job_name": job_name,
"command": command,
"environment_variables": environment_variables,
}
)
base_operator_args = set(inspect.signature(CloudRunExecuteJobOperator.__init__).parameters.keys())
base_kwargs = {}
for arg_key, arg_value in kwargs.items():
if arg_key in base_operator_args:
base_kwargs[arg_key] = arg_value
base_kwargs["task_id"] = kwargs["task_id"]
CloudRunExecuteJobOperator.__init__(self, **base_kwargs)



def fetch_remote_logs(self, result: Any) -> Any:
"""
Fetch Google Cloud Run job logs from Google Cloud Logging to Airflow logs
The function returns the a list of the log messages

result: result from build and run command in Cloud Run Job
"""

self.log.info("Attempt to retrieve logs from Google Cloud Logging")

execution_name = result["latest_created_execution"]["name"]
execution_time = result["latest_created_execution"]["create_time"]
filter_ = f'resource.type = "cloud_run_job" AND resource.labels.job_name = "{self.job_name}" AND timestamp>="{execution_time}"'

time.sleep(5) # Add sleep time to make sure all the job logs are available when we do the request

try:
client = logging.Client(project=self.project_id)
# Fetch logs associated with the job_name
entries = client.list_entries(filter_=filter_)
self.log.info(f"Listing logs of the execution {execution_name} at {execution_time}:")

# List to store log messages
log_messages = []

for entry in entries:
# Search for logs associated with the job executed
if entry.labels["run.googleapis.com/execution_name"] == execution_name:
log_messages.append(entry.payload)
self.log.info(f"Cloud Run Log: {entry.payload}")

return log_messages

except GoogleCloudError as e:
# Catch Google Cloud-related errors (e.g., permission issues)
self.log.warning(f"Warning: Error retrieving logs from Google Cloud Logging: {str(e)}")
return
# Continue without raising an error, just log the warning


And what I tried to do so far as test:

@skip_on_empty_operator
@patch("google.cloud.logging.Client.list_entries")
def test_dbt_gcp_cloud_run_job_fetch_clean_logs():
"""
Check that fetch remote logs run correctly.
"""
run_operator = DbtRunGcpCloudRunJobOperator(**BASE_KWARGS)
mock_list_entries = {
"latest_created_execution": {"name": "test-job-cloud-run", "create_time": "2025-03-19T09:41:22.608063Z"}
}
log_messages = run_operator.fetch_remote_logs(mock_list_entries)

assert len(log_messages) > 0


I know sometimes I don't use the correct words to express what I do/use, so please feel free to correct me if I don't express myself correctly.

Thank you in advance for any suggestion you could give me.

Continue reading...