| """Tests for tests.protoascii configs.""" |
| |
| import abc |
| import collections |
| import itertools |
| import os |
| import re |
| import unittest |
| |
| from catatester.genfiles import continuous_tests_pb2 |
| from catatester import config_utils |
| |
| # Import from eureka-internal/builder/masters project. |
| # pylint: disable=ungrouped-imports |
| try: |
| from src.genfiles import build_config_pb2 |
| except ImportError: |
| from catatester.genfiles import build_config_pb2 |
| |
| import google.protobuf.text_format |
| |
| _CATATESTER_ROOT_PATH = os.path.dirname(os.path.abspath(__file__)) |
| _CONTINUOUS_TESTS_ROOT_PATH = os.path.dirname(_CATATESTER_ROOT_PATH) |
| |
| _SKIP_UNIT_TEST_RE = re.compile(r'unit_test.*') |
| _REQUIRED_FIELDS = ['TEST_CASE_CLASS', 'TEST_CASE_MODULE', |
| 'TEST_SEQUENCE_CLASS', 'TEST_SEQUENCE_MODULE'] |
| |
| # Cap the execution_timeout to 23 hours (b/176174164) |
| _MAX_EXECUTION_TIMEOUT_MINS = 23 * 60 |
| |
| # Assumes full test manifest repo checkout. |
| _BUILD_CONFIG_FILEPATH = os.path.join( |
| os.path.dirname(_CONTINUOUS_TESTS_ROOT_PATH), 'builder', 'masters', |
| 'configs', 'eureka', 'build_config.protoascii') |
| |
| |
| def _GetBuildConfig(): |
| """Returns the build_config protoascii from the builder/masters project. |
| |
| Returns: |
| Ingested build_config. |
| """ |
| build_config = build_config_pb2.BuildConfig() |
| with open(_BUILD_CONFIG_FILEPATH) as f: |
| google.protobuf.text_format.Merge(f.read(), build_config) |
| return build_config |
| |
| |
| # pylint: disable=no-member |
| # Test is not directly derived from UnitTest base class. |
| class ConfigTest(object): |
| """A class to test a continuous_tests config. |
| |
| Note: This class will be tested twice, once with the staging config |
| and once with the prod config. |
| """ |
| ASSISTANT = build_config_pb2.OtaBuild.Capability.Value( |
| 'ASSISTANT') |
| CAST_VIDEO_RECEIVER = build_config_pb2.OtaBuild.Capability.Value( |
| 'CAST_VIDEO_RECEIVER') |
| CAST_AUDIO_RECEIVER = build_config_pb2.OtaBuild.Capability.Value( |
| 'CAST_AUDIO_RECEIVER') |
| |
| def __init__(self): |
| self.__metaclass__ = abc.ABCMeta |
| |
| @abc.abstractproperty |
| def tests_proto(self): |
| pass |
| |
| @property |
| def build_config_proto(self): |
| return _GetBuildConfig() |
| |
| @property |
| def all_tests(self): |
| return ( |
| list(self.tests_proto.continuous_test) + |
| list(self.tests_proto.scheduled_test)) |
| |
| def testContinuousTestsConfig_IsValidProto(self): |
| """Tests the continuous_tests config is a valid proto.""" |
| self.assertTrue(self.tests_proto) |
| |
| def testEveryTestTarget_hasTestConfigOrTestBinary(self): |
| """Test that every test_target has a test_config or test_binary_name.""" |
| for test in self.all_tests: |
| self.assertTrue( |
| test.test_target.binary_name or |
| test.test_target.config) |
| |
| def testEveryScheduledTestsBuildTargetsHasProductsOrProductTypes(self): |
| """Test that scheduled test build_target has products or product types.""" |
| for test in self.tests_proto.scheduled_test: |
| for target in test.build_targets: |
| msg = ('Scheduled tests require build_target.products or' |
| 'build_target.product_types: %s' % test) |
| self.assertTrue( |
| target.products or |
| target.product_types, msg) |
| |
| def testEveryScheduledTestHasSetACron(self): |
| """Verify every scheduled test has set a cron string.""" |
| for test in self.tests_proto.scheduled_test: |
| self.assertTrue(test.schedule.cron, |
| 'scheduled_test is missing cron string: %s' % test) |
| |
| def testEveryTestTargetGroupIsDefined(self): |
| """Test that all test target groups are defined in test_group_tags.""" |
| for test in self.all_tests: |
| for test_group in test.test_target.test_groups: |
| self.assertIn(test_group, self.tests_proto.test_group_tags) |
| |
| def testEveryTestContainsGroup(self): |
| """Ensures group is included in every test.""" |
| for test in self.all_tests: |
| groups = [dim for dim in test.test_target.testbed_requirements if |
| dim.bot_dimension.lower() == 'group'] |
| self.assertTrue(groups, |
| 'Missing group in target: %s' % test.test_target.config) |
| |
| def testEveryTestHasValidTestTimeout(self): |
| """Ensures test_timeout_mins is under the max allowed value.""" |
| for test in self.all_tests: |
| self.assertLessEqual( |
| test.test_target.test_timeout_mins, |
| _MAX_EXECUTION_TIMEOUT_MINS, |
| 'Timeout mins > %s in target: %s' % (_MAX_EXECUTION_TIMEOUT_MINS, |
| test.test_target.config)) |
| |
| def testEveryTargetedTestConfigFileExists(self): |
| """Test that every targeted test_config file exists.""" |
| catatester_path = os.path.abspath(os.path.dirname(__file__)) |
| root_path = os.path.dirname(catatester_path) |
| test_configs_root = os.path.join( |
| root_path, self.tests_proto.test_configs_root) |
| for test in self.all_tests: |
| if test.test_target.config: |
| config_abspath = os.path.join( |
| test_configs_root, |
| test.test_target.config) |
| self.assertTrue( |
| os.path.isfile(config_abspath), |
| 'test_target.config does not exist: %s' % config_abspath) |
| |
| def testEveryTestConfigFileContainsRequiredFields(self): |
| """Test that every test_config file contains required fields.""" |
| |
| for test in self.all_tests: |
| if (not test.test_target.config |
| or re.match(_SKIP_UNIT_TEST_RE, test.test_target.config)): |
| continue |
| |
| fname = os.path.join(_CONTINUOUS_TESTS_ROOT_PATH, |
| self.tests_proto.test_configs_root, test.test_target.config) |
| with open(fname) as f: |
| lines = f.readlines() |
| |
| found_fields = [] |
| for field in _REQUIRED_FIELDS: |
| for line in lines: |
| if not re.match('^%s.*' % field, line, re.I): |
| continue |
| _, value = re.split('=', line, 2) |
| self.assertIsNotNone( value.strip(), |
| '%s in %s must have a value.' % (field, fname)) |
| found_fields.append(field) |
| |
| self.assertEqual(_REQUIRED_FIELDS, sorted(found_fields), |
| '%s must contain all of %s.' % (fname, _REQUIRED_FIELDS)) |
| |
| def testTarget_ValidCipdRefsKey(self): |
| """Tests targets provide valid cipd refs.""" |
| cipd_ref_keys = self.tests_proto.cipd_refs.keys() |
| for test in self.all_tests: |
| msg = 'cipd_ref_key should be one of %s but got %s' % ( |
| cipd_ref_keys, test.test_target.cipd_refs_key) |
| self.assertIn(test.test_target.cipd_refs_key, cipd_ref_keys, msg) |
| |
| def testContinuousTestsConfigDuplicates(self): |
| """Tests there are no duplicated definitions for continuous_tests.""" |
| dict_continuous_tests = collections.defaultdict(list) |
| for continuous_test in self.tests_proto.continuous_test: |
| products_set = self._GetProductsForContinuousTest(continuous_test) |
| config_variants = list(continuous_test.build_targets)[0].variants |
| testbed_reqs = self._GetTestbedRequirementsForContinuousTest( |
| continuous_test) |
| config_key = '%s-%s-%s' % ( |
| continuous_test.test_target.config, |
| continuous_tests_pb2.BuildTargets.Variant.Name(config_variants), |
| ';'.join(testbed_reqs)) |
| if not continuous_test.test_target.test_cases: |
| dict_continuous_tests[config_key].append(products_set) |
| else: |
| for test_case in continuous_test.test_target.test_cases: |
| config_test_case_key = '%s-%s' % (config_key, test_case) |
| dict_continuous_tests[config_test_case_key].append(products_set) |
| for config in dict_continuous_tests: |
| combinations = itertools.combinations(dict_continuous_tests[config], 2) |
| duplicates = next((c for c in combinations if set.intersection(*c)), None) |
| msg = 'Overlapping config found for %s, products %s' % (config, |
| duplicates) |
| self.assertIsNone(duplicates, msg) |
| |
| def _GetDevicesByCapability(self, capability_index): |
| """Getting sets of devices filtered by the index received. |
| |
| Args: |
| capability_index: Number provided to filter the capabilities of devices |
| |
| Returns: |
| returns the set of filtered devices |
| For example, if capability_index is CAST_VIDEO_RECEIVER: |
| {'chorizo', 'steak', 'salami'} |
| """ |
| ota_dict = self.build_config_proto.ota_build |
| set_cast_products = set() |
| for key in ota_dict: |
| if ota_dict[key].capabilities: |
| if capability_index in list(ota_dict[key].capabilities): |
| set_cast_products.add(ota_dict[key].product) |
| return set_cast_products |
| |
| def _GetProductsForContinuousTest(self, continuous_test): |
| """Getting set of products for the specified continuous_test. |
| |
| Args: |
| continuous_test: continuous test object. |
| |
| Returns: |
| returns a set with products of the continuous test |
| """ |
| assistant_devices = self._GetDevicesByCapability(self.ASSISTANT) |
| cast_video_receiver = self._GetDevicesByCapability( |
| self.CAST_VIDEO_RECEIVER) |
| cast_audio_receiver = self._GetDevicesByCapability( |
| self.CAST_AUDIO_RECEIVER) |
| set_products = set() |
| set_product_types = set() |
| for build_target in continuous_test.build_targets: |
| if build_target.products: |
| set_products.update(list(build_target.products)) |
| if build_target.product_types: |
| product_types = list(build_target.product_types) |
| if 'ASSISTANT' in product_types: |
| set_product_types |= assistant_devices |
| if 'CAST_AUDIO_RECEIVER' in product_types: |
| set_product_types |= cast_audio_receiver |
| if 'CAST_VIDEO_RECEIVER' in product_types: |
| set_product_types |= cast_video_receiver |
| products = set_products | set_product_types |
| return products |
| |
| def _GetTestbedRequirementsForContinuousTest(self, continuous_test): |
| """Returns a list of bot dimensions required for the test.""" |
| test_target = continuous_test.test_target |
| if not test_target.skip_default_dimensions: |
| return ['dims:default'] |
| requirements = [] |
| for req in test_target.testbed_requirements: |
| requirements.append(':'.join([req.bot_dimension, req.value])) |
| return requirements |
| |
| def testEveryTestTargetBinaryNameExists(self): |
| """Test that every test_target.binary_name does exist as subdir. |
| |
| Except for test_target with recipe as recipe defines subdir. |
| """ |
| binaries_and_recipes = {(test.test_target.binary_name, |
| test.test_target.recipe) |
| for test in self.all_tests |
| if test.test_target.binary_name} |
| for binary_name, recipe in binaries_and_recipes: |
| subdir = os.path.join(_CONTINUOUS_TESTS_ROOT_PATH, |
| 'lab_system', 'master', 'config', 'tests', |
| binary_name) |
| subdir = os.path.abspath(subdir) |
| self.assertTrue(os.path.isdir(subdir) or recipe, |
| msg='%s does not exist.' % subdir) |
| |
| |
| class ProdConfigTest(unittest.TestCase, ConfigTest): |
| |
| @property |
| def tests_proto(self): |
| return config_utils.parse_continuous_tests_protoascii( |
| config_utils.PROD_CONFIG_PATH) |
| |
| @property |
| def build_restrictions(self): |
| return self.tests_proto.build_restrictions |
| |
| |
| class StagingConfigTest(unittest.TestCase, ConfigTest): |
| |
| @property |
| def tests_proto(self): |
| return config_utils.parse_continuous_tests_protoascii( |
| config_utils.STAGING_CONFIG_PATH) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
