Add unit tests for GcpBodyFieldValidator in google cloud providers (#10003)
This commit is contained in:
Shekhar Singh
GitHub
Родитель
cb5430d2bd
Коммит
0142abb198
|
|
@ -0,0 +1,293 @@
|
|||
# Licensed to the Apache Software Foundation (ASF) under one
|
||||
# or more contributor license agreements. See the NOTICE file
|
||||
# distributed with this work for additional information
|
||||
# regarding copyright ownership. The ASF licenses this file
|
||||
# to you under the Apache License, Version 2.0 (the
|
||||
# "License"); you may not use this file except in compliance
|
||||
# with the License. You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing,
|
||||
# software distributed under the License is distributed on an
|
||||
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
# KIND, either express or implied. See the License for the
|
||||
# specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import unittest
|
||||
|
||||
from airflow.providers.google.cloud.utils.field_validator import (
|
||||
GcpBodyFieldValidator, GcpFieldValidationException, GcpValidationSpecificationException,
|
||||
)
|
||||
|
||||
|
||||
class TestGcpBodyFieldValidator(unittest.TestCase):
|
||||
def test_validate_should_not_raise_exception_if_field_and_body_are_both_empty(self):
|
||||
specification = []
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_if_body_is_none(self):
|
||||
specification = []
|
||||
body = None
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(AttributeError):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_fail_if_specification_is_none(self):
|
||||
specification = None
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_raise_exception_name_attribute_is_missing_from_specs(self):
|
||||
specification = [dict(allow_empty=False)]
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(KeyError):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_raise_exception_if_field_is_not_present(self):
|
||||
specification = [dict(name="name", allow_empty=False)]
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_validate_a_single_field(self):
|
||||
specification = [dict(name="name", allow_empty=False)]
|
||||
body = {"name": "bigquery"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_if_body_is_not_a_dict(self):
|
||||
specification = [dict(name="name", allow_empty=False)]
|
||||
body = [{"name": "bigquery"}]
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(AttributeError):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_fail_for_set_allow_empty_when_field_is_none(self):
|
||||
specification = [dict(name="name", allow_empty=True)]
|
||||
body = {"name": None}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_interpret_allow_empty_clause(self):
|
||||
specification = [dict(name="name", allow_empty=True)]
|
||||
body = {"name": ""}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_raise_if_empty_clause_is_false(self):
|
||||
specification = [dict(name="name", allow_empty=False)]
|
||||
body = {"name": None}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_raise_if_version_mismatch_is_found(self):
|
||||
specification = [dict(name="name", allow_empty=False, api_version='v2')]
|
||||
body = {"name": "value"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_interpret_optional_irrespective_of_allow_empty(self):
|
||||
specification = [dict(name="name", allow_empty=False, optional=True)]
|
||||
body = {"name": None}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_interpret_optional_clause(self):
|
||||
specification = [dict(name="name", allow_empty=False, optional=True)]
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_raise_exception_if_optional_clause_is_false_and_field_not_present(self):
|
||||
specification = [dict(name="name", allow_empty=False, optional=False)]
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_interpret_dict_type(self):
|
||||
specification = [dict(name="labels", optional=True, type="dict")]
|
||||
body = {"labels": {"one": "value"}}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_if_value_is_not_dict_as_per_specs(self):
|
||||
specification = [dict(name="labels", optional=True, type="dict")]
|
||||
body = {"labels": 1}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_not_allow_both_type_and_allow_empty_in_a_spec(self):
|
||||
specification = [dict(name="labels", optional=True, type="dict", allow_empty=True)]
|
||||
body = {"labels": 1}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpValidationSpecificationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_allow_type_and_optional_in_a_spec(self):
|
||||
specification = [dict(name="labels", optional=True, type="dict")]
|
||||
body = {"labels": {}}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_if_union_field_is_not_found(self):
|
||||
specification = [
|
||||
dict(name="an_union", type="union", optional=False, fields=[
|
||||
dict(name="variant_1", regexp=r'^.+$', optional=False, allow_empty=False),
|
||||
])
|
||||
]
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_if_there_is_no_nested_field_for_union(self):
|
||||
specification = [
|
||||
dict(name="an_union", type="union", optional=False, fields=[])
|
||||
]
|
||||
body = {}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
|
||||
with self.assertRaises(GcpValidationSpecificationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_interpret_union_with_one_field(self):
|
||||
specification = [
|
||||
dict(name="an_union", type="union", fields=[
|
||||
dict(name="variant_1", regexp=r'^.+$'),
|
||||
])
|
||||
]
|
||||
body = {"variant_1": "abc", "variant_2": "def"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_if_both_field_of_union_is_present(self):
|
||||
specification = [
|
||||
dict(name="an_union", type="union", fields=[
|
||||
dict(name="variant_1", regexp=r'^.+$'),
|
||||
dict(name="variant_2", regexp=r'^.+$'),
|
||||
])
|
||||
]
|
||||
body = {"variant_1": "abc", "variant_2": "def"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_validate_when_value_matches_regex(self):
|
||||
specification = [
|
||||
dict(name="an_union", type="union", fields=[
|
||||
dict(name="variant_1", regexp=r'[^a-z]'),
|
||||
])
|
||||
]
|
||||
body = {"variant_1": "12"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_fail_when_value_does_not_match_regex(self):
|
||||
specification = [
|
||||
dict(name="an_union", type="union", fields=[
|
||||
dict(name="variant_1", regexp=r'[^a-z]'),
|
||||
])
|
||||
]
|
||||
body = {"variant_1": "abc"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_raise_if_custom_validation_is_not_true(self):
|
||||
def _int_equal_to_zero(value):
|
||||
if int(value) != 0:
|
||||
raise GcpFieldValidationException("The available memory has to be equal to 0")
|
||||
|
||||
specification = [
|
||||
dict(name="availableMemoryMb", custom_validation=_int_equal_to_zero)
|
||||
]
|
||||
body = {"availableMemoryMb": 1}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
with self.assertRaises(GcpFieldValidationException):
|
||||
validator.validate(body)
|
||||
|
||||
def test_validate_should_not_raise_if_custom_validation_is_true(self):
|
||||
def _int_equal_to_zero(value):
|
||||
if int(value) != 0:
|
||||
raise GcpFieldValidationException("The available memory has to be equal to 0")
|
||||
|
||||
specification = [
|
||||
dict(name="availableMemoryMb", custom_validation=_int_equal_to_zero)
|
||||
]
|
||||
body = {"availableMemoryMb": 0}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
self.assertIsNone(validator.validate(body))
|
||||
|
||||
def test_validate_should_validate_group_of_specs(self):
|
||||
specification = [
|
||||
dict(name="name", allow_empty=False),
|
||||
dict(name="description", allow_empty=False, optional=True),
|
||||
dict(name="labels", optional=True, type="dict"),
|
||||
dict(name="an_union", type="union", fields=[
|
||||
dict(name="variant_1", regexp=r'^.+$'),
|
||||
dict(name="variant_2", regexp=r'^.+$', api_version='v1beta2'),
|
||||
dict(name="variant_3", type="dict", fields=[
|
||||
dict(name="url", regexp=r'^.+$')
|
||||
]),
|
||||
dict(name="variant_4")
|
||||
]),
|
||||
]
|
||||
body = {"variant_1": "abc", "name": "bigquery"}
|
||||
|
||||
validator = GcpBodyFieldValidator(specification, 'v1')
|
||||
validator.validate(body)
|
||||
|
|
@ -31,7 +31,6 @@ MISSING_TEST_FILES = {
|
|||
'tests/providers/google/cloud/log/test_gcs_task_handler.py',
|
||||
'tests/providers/google/cloud/operators/test_datastore.py',
|
||||
'tests/providers/google/cloud/utils/test_field_sanitizer.py',
|
||||
'tests/providers/google/cloud/utils/test_field_validator.py',
|
||||
'tests/providers/google/cloud/utils/test_mlengine_prediction_summary.py',
|
||||
'tests/providers/microsoft/azure/sensors/test_azure_cosmos.py',
|
||||
'tests/providers/microsoft/azure/log/test_wasb_task_handler.py',
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче