Блог

Это одна из функций в моем скрипте python, для которой я пытаюсь написать модульный тестовый пример, поскольку она использует глобальные переменные и функции аудита и больших запросов, которые написаны в виде разных служебных скриптов. Я не понимаю, как написать @patch и выполнить модульные тестовые примеры для одного и того же.

1. Как я буду исправлять глобальные переменные?
2. Как исправить функции, которые не возвращают никаких значений, например: audit_event_source_table, можем ли мы игнорировать такие функции во время модульного тестирования? если да, то как сделать то же самое?
3. Как выполнить утверждение, поскольку у меня нет никакого возвращаемого значения, но есть logger.info сообщения?

 import logging
from datetime import datetime
from pathlib import Path
import sys
import __main__
from intient_research_rdm_common.utils.audit_utils import audit_event_source_table, audit_event_job_table, 
    get_job_id, get_source_object_id

from intient_research_rdm_kg_core.common_utils.utils.bigquery_utils import bigquery_data_read
from intient_research_rdm_kg_core.common_utils.utils.conf_read import read_args, read_source_config, read_env_config


global project_id, service_account, conn_ip, debug, node_table_list, edge_table_list, source_name

def edge_validation():
    global edge_table_list
    global source_name
    edge_table_na = []
    edge_table_list_rowcount_zero = []
    dataset_e = "prep_e_"   source_name
    row_count = 0
    edge_table = ""
    source_object_start_timestamp = datetime.now()
    source_object_id = get_source_object_id(source_name, source_object_start_timestamp)
    source_object_type = AUDIT_SOURCE_OBJECT_TYPE_BIGQUERY
    job_id = get_job_id(source_object_start_timestamp)
    source_object_name = dataset_e
    try:
        for edge_table in edge_table_list:
            sql_query = " SELECT * FROM "   "`"   project_id   "."   dataset_e   ".__TABLES__` WHERE table_id ="   "'"   edge_table   "'"
            data_read, col_names = bigquery_data_read(service_account, sql_query, project_id)
            for ind in data_read.index:
                row_count = (data_read['row_count'][ind])
            if len(data_read.index) == 0:
                edge_table_na.append(edge_table)
            elif row_count == 0:
                edge_table_list_rowcount_zero.append(edge_table)

        if len(edge_table_na) > 0:
            logging.info("Missing Edge tables in preprocessing layer {} ".format(edge_table_na))
        if len(edge_table_list_rowcount_zero) > 0:
            logging.info("Edge tables with row count as zero in Pre-processing layer  {} ".format(edge_table_list_rowcount_zero))
        if len(edge_table_na) == 0 and len(edge_table_list_rowcount_zero) == 0:
            logging.info(
                "Edge list validation for the source {} has been successfully completed with no discrepancies".format(
                    source_name))
            audit_event_source_table(source_object_id, source_object_name, source_object_type, source_name,
                                     source_object_name,
                                     job_id, AUDIT_JOB_STATUS_PASS, source_object_start_timestamp,
                                     datetime.now(), 'NA', 'NA', project_id)

        if len(edge_table_na) > 0 or len(edge_table_list_rowcount_zero) > 0:
            audit_event_source_table(source_object_id, source_object_name, source_object_type, source_name,
                                     source_object_name,
                                     job_id, AUDIT_JOB_STATUS_PASS, source_object_start_timestamp,
                                     datetime.now(), 'NA', 'NA', project_id)
            sys.exit(1)
    except Exception as e:
        msg = '{} : Issue with the edge validation for {} is: n{}n'.format(datetime.now(), edge_table, e)
        logging.error(msg)
        audit_event_source_table(source_object_id, source_object_name, source_object_type, source_name,
                                 source_object_name,
                                 job_id, AUDIT_JOB_STATUS_FAIL, source_object_start_timestamp,
                                 datetime.now(), AUDIT_ERROR_TYPE_PREPROCESSING_KG_LAYER_VALIDATION, msg,
                                 project_id)
        raise Exception(msg)
  

1. Исправление глобальных переменных — таким же образом, как вы исправляете метод класса, вы исправляете глобальную переменную. В вашем фрагменте кода неясно, где определены глобальные переменные (т.Е. вы импортируете эти переменные из другого модуля или присваиваете этим переменным в верхней части вашего скрипта Python). В любом случае, вы исправляете пространство имен, в котором используется функция. Если вы сможете подтвердить дополнительные детали, я смогу помочь.
2. Лично я исправляю и тестирую функции без возвращаемого значения одинаково. Например, если бы я хотел исправить переменную source_object_start_timestamp, я бы использовал: source_object_start_timestamp = patch('pandas.datetime.utcnow', return_value="2020-08-16 20:36:06.578174").start() . Для функций BigQuery я бы все равно исправил их, но в вашем модульном тестировании используйте метод mock_call_count класса unittest.mock.mock, чтобы проверить, была ли вызвана эта функция.
3. Пункт 2 адресует ваш третий запрос — используйте метод mock_call_count, чтобы проверить, сколько раз был вызван макет