# Copyright 2024 The Pigweed Authors # # Licensed 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 # # https://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. """Unit tests for the sensor metadata validator""" from pathlib import Path import unittest import tempfile import yaml from pw_sensor.validator import Validator class ValidatorTest(unittest.TestCase): """Tests the Validator class.""" maxDiff = None def test_missing_compatible(self) -> None: """Check that missing 'compatible' key throws exception""" self._check_with_exception( metadata={}, exception_string="ERROR: Malformed sensor metadata YAML:\n{}", cause_substrings=["'compatible' is a required property"], ) def test_invalid_compatible_type(self) -> None: """Check that incorrect type of 'compatible' throws exception""" self._check_with_exception( metadata={"compatible": {}, "supported-buses": ["i2c"]}, exception_string=( "ERROR: Malformed sensor metadata YAML:\ncompatible: {}\n" + "supported-buses:\n- i2c" ), cause_substrings=[ "'part' is a required property", ], ) self._check_with_exception( metadata={"compatible": [], "supported-buses": ["i2c"]}, exception_string=( "ERROR: Malformed sensor metadata YAML:\ncompatible: []\n" + "supported-buses:\n- i2c" ), cause_substrings=["[] is not of type 'object'"], ) self._check_with_exception( metadata={"compatible": 1, "supported-buses": ["i2c"]}, exception_string=( "ERROR: Malformed sensor metadata YAML:\ncompatible: 1\n" + "supported-buses:\n- i2c" ), cause_substrings=["1 is not of type 'object'"], ) self._check_with_exception( metadata={"compatible": "", "supported-buses": ["i2c"]}, exception_string=( "ERROR: Malformed sensor metadata YAML:\ncompatible: ''\n" + "supported-buses:\n- i2c" ), cause_substrings=[" is not of type 'object'"], ) def test_partial_compatible_string(self) -> None: """ Check that missing 'org' generates correct keys and empty entries are removed. """ metadata: dict = { "compatible": {"part": "pigweed"}, "supported-buses": ["i2c"], } result = Validator().validate(metadata=metadata) self.assertIn("pigweed", result["sensors"]) self.assertDictEqual( {"part": "pigweed"}, result["sensors"]["pigweed"]["compatible"], ) metadata["compatible"]["org"] = " " result = Validator().validate(metadata=metadata) self.assertIn("pigweed", result["sensors"]) self.assertDictEqual( {"part": "pigweed"}, result["sensors"]["pigweed"]["compatible"], ) def test_compatible_string_to_lower(self) -> None: """ Check that compatible components are converted to lowercase and stripped. """ metadata = { "compatible": {"org": "Google", "part": "Pigweed"}, "supported-buses": ["i2c"], } result = Validator().validate(metadata=metadata) self.assertIn("google,pigweed", result["sensors"]) self.assertDictEqual( {"org": "google", "part": "pigweed"}, result["sensors"]["google,pigweed"]["compatible"], ) def test_invalid_supported_buses(self) -> None: """ Check that invalid or missing supported-buses cause an error """ self._check_with_exception( metadata={"compatible": {"org": "Google", "part": "Pigweed"}}, exception_string=( "ERROR: Malformed sensor metadata YAML:\ncompatible:\n" + " org: Google\n part: Pigweed" ), cause_substrings=[], ) self._check_with_exception( metadata={ "compatible": {"org": "Google", "part": "Pigweed"}, "supported-buses": [], }, exception_string=( "ERROR: Malformed sensor metadata YAML:\ncompatible:\n" + " org: Google\n part: Pigweed\nsupported-buses: []" ), cause_substrings=[], ) def test_unique_bus_names(self) -> None: """ Check that resulting bus names are unique and are converted to lowercase """ self._check_with_exception( metadata={ "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c", "I2C", "SPI"], "deps": [], }, exception_string=( "ERROR: bus list contains duplicates when converted to " "lowercase and concatenated with '_': " "['I2C', 'SPI', 'i2c'] -> ['i2c', 'spi']" ), cause_substrings=[], ) self._check_with_exception( metadata={ "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i 2 c", "i 2_c", "i\t2-c"], "deps": [], }, exception_string=( "ERROR: bus list contains duplicates when converted to " "lowercase and concatenated with '_': " "['i\\t2-c', 'i 2_c', 'i 2 c'] -> ['i_2_c']" ), cause_substrings=[], ) def test_invalid_sensor_attribute(self) -> None: attribute = { "attribute": "sample_rate", "channel": "laundry", "trigger": "data_ready", "units": "rate", } dep_filename = self._generate_dependency_file() self._check_with_exception( metadata={ "compatible": {"part": "foo"}, "supported-buses": ["i2c"], "deps": [str(dep_filename.resolve())], "attributes": [attribute], }, exception_string=( "Attribute instances cannot specify both channel AND trigger:\n" + yaml.safe_dump(attribute, indent=2) ), cause_substrings=[], ) def test_empty_dependency_list(self) -> None: """ Check that an empty or missing 'deps' resolves to one with an empty 'deps' list """ expected = { "sensors": { "google,foo": { "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c"], "description": "", "channels": {}, "attributes": [], "triggers": [], "extras": {}, }, }, "channels": {}, "attributes": {}, "triggers": {}, "units": {}, } metadata = { "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c"], "deps": [], } result = Validator().validate(metadata=metadata) self.assertEqual(result, expected) metadata = { "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c"], } result = Validator().validate(metadata=metadata) self.assertEqual(result, expected) def test_invalid_dependency_file(self) -> None: """ Check that if an invalid dependency file is listed, we throw an error. We know this will not be a valid file, because we have no files in the include path so we have nowhere to look for the file. """ self._check_with_exception( metadata={ "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c"], "deps": ["test.yaml"], }, exception_string="Failed to find test.yaml using search paths:", cause_substrings=[], exception_type=FileNotFoundError, ) def test_invalid_channel_name_raises_exception(self) -> None: """ Check that if given a channel name that's not defined, we raise an Error """ self._check_with_exception( metadata={ "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c"], "channels": {"bar": []}, }, exception_string="Failed to find a definition for 'bar', did" " you forget a dependency?", cause_substrings=[], ) @staticmethod def _generate_dependency_file() -> Path: with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", encoding="utf-8", delete=False ) as dep: dep_filename = Path(dep.name) dep.write( yaml.safe_dump( { "units": { "rate": { "symbol": "Hz", }, "sandwiches": { "symbol": "sandwiches", }, "squeaks": {"symbol": "squeaks"}, "items": { "symbol": "items", }, }, "attributes": { "sample_rate": {}, }, "channels": { "bar": { "units": "sandwiches", }, "soap": { "name": "The soap", "description": ( "Measurement of how clean something is" ), "units": "squeaks", }, "laundry": { "description": "Clean clothes count", "units": "items", }, }, "triggers": { "data_ready": { "description": "notify when new data is ready", }, }, }, ) ) return dep_filename def test_channel_info_from_deps(self) -> None: """ End to end test resolving a dependency file and setting the right default attribute values. """ dep_filename = self._generate_dependency_file() metadata = Validator(include_paths=[dep_filename.parent]).validate( metadata={ "compatible": {"org": "google", "part": "foo"}, "supported-buses": ["i2c"], "deps": [dep_filename.name], "attributes": [ # Attribute applied to a channel { "attribute": "sample_rate", "channel": "laundry", "units": "rate", }, # Attribute applied to the entire device { "attribute": "sample_rate", "units": "rate", }, # Attribute applied to a trigger { "attribute": "sample_rate", "trigger": "data_ready", "units": "rate", }, ], "channels": { "bar": [], "soap": [], "laundry": [ {"name": "kids' laundry"}, {"name": "adults' laundry"}, ], }, "triggers": [ "data_ready", ], }, ) # Check attributes self.assertEqual( metadata, { "attributes": { "sample_rate": { "name": "sample_rate", "description": "", }, }, "channels": { "bar": { "name": "bar", "description": "", "units": "sandwiches", }, "soap": { "name": "The soap", "description": "Measurement of how clean something is", "units": "squeaks", }, "laundry": { "name": "laundry", "description": "Clean clothes count", "units": "items", }, }, "triggers": { "data_ready": { "name": "data_ready", "description": "notify when new data is ready", }, }, "units": { "rate": { "name": "Hz", "symbol": "Hz", "description": "", }, "sandwiches": { "name": "sandwiches", "symbol": "sandwiches", "description": "", }, "squeaks": { "name": "squeaks", "symbol": "squeaks", "description": "", }, "items": { "name": "items", "symbol": "items", "description": "", }, }, "sensors": { "google,foo": { "description": "", "compatible": { "org": "google", "part": "foo", }, "supported-buses": ["i2c"], "attributes": [ { "attribute": "sample_rate", "channel": "laundry", "units": "rate", }, { "attribute": "sample_rate", "units": "rate", }, { "attribute": "sample_rate", "trigger": "data_ready", "units": "rate", }, ], "channels": { "bar": [ { "name": "bar", "description": "", "units": "sandwiches", }, ], "soap": [ { "name": "The soap", "description": ( "Measurement of how clean something is" ), "units": "squeaks", }, ], "laundry": [ { "name": "kids' laundry", "description": "Clean clothes count", "units": "items", }, { "name": "adults' laundry", "description": "Clean clothes count", "units": "items", }, ], }, "triggers": ["data_ready"], "extras": {}, }, }, }, ) def _check_with_exception( self, metadata: dict, exception_string: str, cause_substrings: list[str], exception_type: type[BaseException] = RuntimeError, ) -> None: with self.assertRaises(exception_type) as context: Validator().validate(metadata=metadata) self.assertEqual( str(context.exception).rstrip(), str(exception_string).rstrip() ) for cause_substring in cause_substrings: self.assertTrue( cause_substring in str(context.exception.__cause__), f"Actual cause: {str(context.exception.__cause__)}", ) if __name__ == "__main__": unittest.main()