@mock.patch('airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.create_empty_table')
def test_create_external_table_with_kms(self, mock_create):
    external_project_dataset_table = f'{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}'
    source_uris = ['test_data.csv']
    source_format = 'CSV'
    autodetect = False
    compression = 'NONE'
    ignore_unknown_values = False
    max_bad_records = 10
    skip_leading_rows = 1
    field_delimiter = ','
    quote_character = None
    allow_quoted_newlines = False
    allow_jagged_rows = False
    encoding = 'UTF-8'
    labels = {'label1': 'test1', 'label2': 'test2'}
    schema_fields = [{'mode': 'REQUIRED', 'name': 'id', 'type': 'STRING', 'description': None}]
    encryption_configuration = {'kms_key_name': 'projects/p/locations/l/keyRings/k/cryptoKeys/c'}
    self.hook.create_external_table(external_project_dataset_table=external_project_dataset_table, source_uris=source_uris, source_format=source_format, autodetect=autodetect, compression=compression, ignore_unknown_values=ignore_unknown_values, max_bad_records=max_bad_records, skip_leading_rows=skip_leading_rows, field_delimiter=field_delimiter, quote_character=quote_character, allow_jagged_rows=allow_jagged_rows, encoding=encoding, allow_quoted_newlines=allow_quoted_newlines, labels=labels, schema_fields=schema_fields, encryption_configuration=encryption_configuration)
    body = {'externalDataConfiguration': {'autodetect': autodetect, 'sourceFormat': source_format, 'sourceUris': source_uris, 'compression': compression, 'ignoreUnknownValues': ignore_unknown_values, 'schema': {'fields': schema_fields}, 'maxBadRecords': max_bad_records, 'csvOptions': {'skipLeadingRows': skip_leading_rows, 'fieldDelimiter': field_delimiter, 'quote': quote_character, 'allowQuotedNewlines': allow_quoted_newlines, 'allowJaggedRows': allow_jagged_rows, 'encoding': encoding}}, 'tableReference': {'projectId': PROJECT_ID, 'datasetId': DATASET_ID, 'tableId': TABLE_ID}, 'labels': labels, 'encryptionConfiguration': encryption_configuration}
    mock_create.assert_called_once_with(table_resource=body, project_id=PROJECT_ID, location=None, exists_ok=True)
@mock.patch('airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.create_empty_table')
def test_create_external_table_with_kms(self, mock_create):
    external_project_dataset_table = f'{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}'
    source_uris = ['test_data.csv']
    source_format = 'CSV'
    autodetect = False
    compression = 'NONE'
    ignore_unknown_values = False
    max_bad_records = 10
    skip_leading_rows = 1
    field_delimiter = ','
    quote_character = None
    allow_quoted_newlines = False
    allow_jagged_rows = False
    encoding = 'UTF-8'
    labels = {'label1': 'test1', 'label2': 'test2'}
    schema_fields = [{'mode': 'REQUIRED', 'name': 'id', 'type': 'STRING', 'description': None}]
    encryption_configuration = {'kms_key_name': 'projects/p/locations/l/keyRings/k/cryptoKeys/c'}
    self.hook.create_external_table(external_project_dataset_table=external_project_dataset_table, source_uris=source_uris, source_format=source_format, autodetect=autodetect, compression=compression, ignore_unknown_values=ignore_unknown_values, max_bad_records=max_bad_records, skip_leading_rows=skip_leading_rows, field_delimiter=field_delimiter, quote_character=quote_character, allow_jagged_rows=allow_jagged_rows, encoding=encoding, allow_quoted_newlines=allow_quoted_newlines, labels=labels, schema_fields=schema_fields, encryption_configuration=encryption_configuration)
    body = {'externalDataConfiguration': {'autodetect': autodetect, 'sourceFormat': source_format, 'sourceUris': source_uris, 'compression': compression, 'ignoreUnknownValues': ignore_unknown_values, 'schema': {'fields': schema_fields}, 'maxBadRecords': max_bad_records, 'csvOptions': {'skipLeadingRows': skip_leading_rows, 'fieldDelimiter': field_delimiter, 'quote': quote_character, 'allowQuotedNewlines': allow_quoted_newlines, 'allowJaggedRows': allow_jagged_rows, 'encoding': encoding}}, 'tableReference': {'projectId': PROJECT_ID, 'datasetId': DATASET_ID, 'tableId': TABLE_ID}, 'labels': labels, 'encryptionConfiguration': encryption_configuration}
    mock_create.assert_called_once_with(table_resource=body, project_id=PROJECT_ID, location=None, exists_ok=True)