Измешивание функции http.get не работает в Pytest ⇐ Python

[Anonymous]

Тестовый файл находится в Status_update_job/tests , и sync_mdl_status.py в папке Status_update/lib/.

Код: Выделить всё

from lib.sync_mdl_status import update_mdl_status

class TestStatusUpdaterMdl(unittest.TestCase):
def setUp(self):
self.postgresql = testing.postgresql.Postgresql()
db_init(self.postgresql)

def tearDown(self):
db_ops.close_session()
self.postgresql.stop()

def test_status_updater_mdl(self):
mock_mdl_http_get = patch('lib.sync_mdl_status.http.get').start()
mock_mdl_http_get.return_value.status_code = 200

batch = update_mdl_status(self.postgresql.url(), "test", 1, aws_auth_token)
< /code>
Файл Update_mdl_status содержит-< /p>
import json
import time
from functools import partial
import multiprocessing
from requests.adapters import HTTPAdapter
from urllib3 import Retry
import db_ops.main as db_operations
import logging as log
import pandas as pd
import numpy as np
import requests
import pandas
from lib.utils import execute_batch_query, is_valid_http_status_code
from scripts.aws_secrets import fetch_aws_secret

http = requests.Session()
http.mount("http://", TimeoutHTTPAdapter(max_retries=HTTP_RETRY_STRATEGY))
http.mount("https://", TimeoutHTTPAdapter(max_retries=HTTP_RETRY_STRATEGY))

def fetch_mdl_token(mdl_auth_aws_secret_name, region: str):
mdl_auth_tokens = fetch_aws_secret(mdl_auth_aws_secret_name, region)
return mdl_auth_tokens

def fetch_mdl_status(mdl_auth_tokens: dict, mdl_endpoint: str, row: dict):
mdl_id = row['mdl_id']
print(mdl_auth_tokens)
mdl_auth_token_to_use = mdl_auth_tokens[row['obfuscate_info']['db_name']]
mdl_headers = {'Content-Type': 'application/json', 'Authorization': 'MDL-AUTH ' + mdl_auth_token_to_use}
res = http.get(f"{mdl_endpoint}/{mdl_id}", headers=mdl_headers)

def process_mdl_batch(batch: pandas.DataFrame, mdl_endpoint: str, mdl_auth_tokens: dict):
batch['status'] = batch['status'].apply(lambda x: x.strip())
batch['new_status'] = batch['status']
return batch.apply(lambda row: fetch_mdl_status(mdl_auth_tokens, mdl_endpoint, row), axis=1)

def update_mdl_status(mdl_con_url: str, schema: str, mdl_threads: int, mdl_auth_tokens: dict):
start = time.time()
log.info("Getting MDL Token and endpoint details")

log.info("Getting one batch_id to fetch status from MDL")
batch_info = db_operations.db.get_a_batch_to_process_its_status()
print(batch_info)

if batch_info is not None:
batch_id = list(batch_info.keys())[0]
region = batch_info[batch_id]
else:
log.info("There are not pending/obfuscated batch ids to process")
return None

mdl_endpoint_for_current_batch = fetch_mdl_endpoint(region)

log.info(f"Current mdl batch to fetch the status for is {batch_id}")
df = db_operations.db.get_pending_mdl_batch(batch_id)
if df is None:
log.info("No pending/obfuscated MDL requests in batch_table_mapping.")
return batch_id

size = len(df.index)
log.debug(f"Retrieving MDL status for {size} batch requests")
# multi thread MDL API calls to fetch request status
batch = np.array_split(df, min(size, mdl_threads))
pool = multiprocessing.Pool(mdl_threads)

res = pd.concat(pool.map(partial(process_mdl_batch, mdl_endpoint=mdl_endpoint_for_current_batch,
mdl_auth_tokens=mdl_auth_tokens), batch))

Намекнув http.get не работает. Я пытался использовать патч и приспособления, но это не работает. Фактический вызов проводится в API вместо использования издевающегося ответа. Я попытался издеваться над lib.sync_mdl_status.requests.session.get , но я получаю ту же ошибку.


Подробнее здесь: https://stackoverflow.com/questions/786 ... -in-pytest