将 GCS 文件加载到 BigQuery 的 Cloud Functions 的 Python 单元测试答案

【问题标题】：Python unit testing for a Cloud Function that loads GCS files to BigQuery将 GCS 文件加载到 BigQuery 的 Cloud Functions 的 Python 单元测试
【发布时间】：2021-03-03 15:27:38
【问题描述】：

这是我第一次使用云函数，这个云函数只做一项工作：每次将文件上传到 GCS 存储桶时，云函数都会运行并将该文件 (.csv) 复制到没有任何转换的 BigQuery 表。测试gcs_to_bq 方法最有效的方法是什么（单元而不是集成）？

def get_bq_table_name(file_name):
    if re.search('car', file_name):
        return 'car'
    return 'bike'

def gcs_to_bq(event, context):

    # Construct a BigQuery client object.
    client = bigquery.Client()

    bq_table = get_bq_table_name(event['name'] )

    table_id = f'xxx.yyy.{bq_table}'
    
    job_config = bigquery.LoadJobConfig(
        schema=[
            bigquery.SchemaField("datetime", "STRING"),
            bigquery.SchemaField("name", "STRING"),
            bigquery.SchemaField("id", "STRING"),

        ],
        skip_leading_rows=1,
        # The source format defaults to CSV, so the line below is optional.
        source_format=bigquery.SourceFormat.CSV,
    )
    
    uri = "gs://" + event['bucket'] + '/' + event['name'] 

    load_job = client.load_table_from_uri(
        uri, table_id, job_config=job_config
    )  # Make an API request.

    load_job.result()  # Waits for the job to complete.

    destination_table = client.get_table(table_id)  # Make an API request.
    print("Loaded {} rows.".format(destination_table.num_rows))

【问题讨论】：

标签： python google-bigquery google-cloud-functions

【解决方案1】：

我认为单元测试需要三样东西

创建一个伪造的.cvs 文件并将其上传到stage/dev GCS 存储桶。
在 BQ 中创建暂存数据集。
创建一个呈现 (1) 的假事件对象。

那么你的单元测试就是用 (3) 调用gcs_to_bq() 并检查表是否在 (2) 中正确创建。

如您所见，虽然它是单元测试，但它需要设置云资源。如果您想完全在本地创建 GCS 存根/模拟，有一个 GCS 模拟器可以提供帮助，但我从未尝试过。 https://github.com/fsouza/fake-gcs-server

【讨论】：