| # 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": {}}, |
| exception_string=( |
| "ERROR: Malformed sensor metadata YAML:\ncompatible: {}" |
| ), |
| cause_substrings=[ |
| "'org' is a required property", |
| ], |
| ) |
| |
| self._check_with_exception( |
| metadata={"compatible": []}, |
| exception_string=( |
| "ERROR: Malformed sensor metadata YAML:\ncompatible: []" |
| ), |
| cause_substrings=["[] is not of type 'object'"], |
| ) |
| |
| self._check_with_exception( |
| metadata={"compatible": 1}, |
| exception_string=( |
| "ERROR: Malformed sensor metadata YAML:\ncompatible: 1" |
| ), |
| cause_substrings=["1 is not of type 'object'"], |
| ) |
| |
| self._check_with_exception( |
| metadata={"compatible": ""}, |
| exception_string=( |
| "ERROR: Malformed sensor metadata YAML:\ncompatible: ''" |
| ), |
| cause_substrings=[" is not of type 'object'"], |
| ) |
| |
| 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"}, |
| "channels": {}, |
| "attributes": {}, |
| "triggers": {}, |
| }, |
| }, |
| "channels": {}, |
| "attributes": {}, |
| "triggers": {}, |
| } |
| metadata = { |
| "compatible": {"org": "google", "part": "foo"}, |
| "deps": [], |
| } |
| result = Validator().validate(metadata=metadata) |
| self.assertEqual(result, expected) |
| |
| metadata = {"compatible": {"org": "google", "part": "foo"}} |
| 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"}, |
| "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"}, |
| "channels": {"bar": {}}, |
| }, |
| exception_string="Failed to find a definition for 'bar', did" |
| " you forget a dependency?", |
| cause_substrings=[], |
| ) |
| |
| def test_channel_info_from_deps(self) -> None: |
| """ |
| End to end test resolving a dependency file and setting the right |
| default attribute values. |
| """ |
| with tempfile.NamedTemporaryFile( |
| mode="w", suffix=".yaml", encoding="utf-8", delete=False |
| ) as dep: |
| dep_filename = Path(dep.name) |
| dep.write( |
| yaml.safe_dump( |
| { |
| "attributes": { |
| "sample_rate": { |
| "units": {"symbol": "Hz"}, |
| }, |
| }, |
| "channels": { |
| "bar": { |
| "units": {"symbol": "sandwiches"}, |
| }, |
| "soap": { |
| "name": "The soap", |
| "description": ( |
| "Measurement of how clean something is" |
| ), |
| "units": {"symbol": "sqeaks"}, |
| }, |
| "laundry": { |
| "description": "Clean clothes count", |
| "units": {"symbol": "items"}, |
| "sub-channels": { |
| "shirts": { |
| "description": "Clean shirt count", |
| }, |
| "pants": { |
| "description": "Clean pants count", |
| }, |
| }, |
| }, |
| }, |
| "triggers": { |
| "data_ready": { |
| "description": "notify when new data is ready", |
| }, |
| }, |
| }, |
| ) |
| ) |
| |
| metadata = Validator(include_paths=[dep_filename.parent]).validate( |
| metadata={ |
| "compatible": {"org": "google", "part": "foo"}, |
| "deps": [dep_filename.name], |
| "attributes": { |
| "sample_rate": {}, |
| }, |
| "channels": { |
| "bar": {}, |
| "soap": { |
| "name": "soap name override", |
| }, |
| "laundry_shirts": {}, |
| "laundry_pants": {}, |
| "laundry": { |
| "indicies": [ |
| {"name": "kids' laundry"}, |
| {"name": "adults' laundry"}, |
| ] |
| }, |
| }, |
| "triggers": { |
| "data_ready": {}, |
| }, |
| }, |
| ) |
| expected_trigger_data_ready = { |
| "name": "data_ready", |
| "description": "notify when new data is ready", |
| } |
| expected_attribute_sample_rate = { |
| "name": "sample_rate", |
| "description": "", |
| "units": {"name": "Hz", "symbol": "Hz"}, |
| } |
| expected_channel_bar = { |
| "name": "bar", |
| "description": "", |
| "units": { |
| "name": "sandwiches", |
| "symbol": "sandwiches", |
| }, |
| } |
| expected_channel_soap = { |
| "name": "The soap", |
| "description": "Measurement of how clean something is", |
| "units": { |
| "name": "sqeaks", |
| "symbol": "sqeaks", |
| }, |
| } |
| expected_channel_laundry_shirts = { |
| "name": "laundry_shirts", |
| "description": "Clean shirt count", |
| "units": { |
| "name": "items", |
| "symbol": "items", |
| }, |
| } |
| expected_channel_laundry_pants = { |
| "name": "laundry_pants", |
| "description": "Clean pants count", |
| "units": { |
| "name": "items", |
| "symbol": "items", |
| }, |
| } |
| expected_channel_laundry = { |
| "name": "laundry", |
| "description": "Clean clothes count", |
| "units": { |
| "name": "items", |
| "symbol": "items", |
| }, |
| } |
| expected_sensor_channel_bar = { |
| "name": "bar", |
| "description": "", |
| "units": { |
| "name": "sandwiches", |
| "symbol": "sandwiches", |
| }, |
| "indicies": [ |
| { |
| "name": "bar", |
| "description": "", |
| }, |
| ], |
| } |
| expected_sensor_channel_soap = { |
| "name": "soap name override", |
| "description": "Measurement of how clean something is", |
| "units": { |
| "name": "sqeaks", |
| "symbol": "sqeaks", |
| }, |
| "indicies": [ |
| { |
| "name": "soap name override", |
| "description": "Measurement of how clean something is", |
| }, |
| ], |
| } |
| expected_sensor_channel_laundry_shirts = { |
| "name": "laundry_shirts", |
| "description": "Clean shirt count", |
| "units": { |
| "name": "items", |
| "symbol": "items", |
| }, |
| "indicies": [ |
| { |
| "name": "laundry_shirts", |
| "description": "Clean shirt count", |
| }, |
| ], |
| } |
| expected_sensor_channel_laundry_pants = { |
| "name": "laundry_pants", |
| "description": "Clean pants count", |
| "units": { |
| "name": "items", |
| "symbol": "items", |
| }, |
| "indicies": [ |
| { |
| "name": "laundry_pants", |
| "description": "Clean pants count", |
| }, |
| ], |
| } |
| expected_sensor_channel_laundry = { |
| "name": "laundry", |
| "description": "Clean clothes count", |
| "units": { |
| "name": "items", |
| "symbol": "items", |
| }, |
| "indicies": [ |
| { |
| "name": "kids' laundry", |
| "description": "Clean clothes count", |
| }, |
| { |
| "name": "adults' laundry", |
| "description": "Clean clothes count", |
| }, |
| ], |
| } |
| self.assertEqual( |
| metadata, |
| { |
| "attributes": {"sample_rate": expected_attribute_sample_rate}, |
| "channels": { |
| "bar": expected_channel_bar, |
| "soap": expected_channel_soap, |
| "laundry_shirts": expected_channel_laundry_shirts, |
| "laundry_pants": expected_channel_laundry_pants, |
| "laundry": expected_channel_laundry, |
| }, |
| "triggers": { |
| "data_ready": expected_trigger_data_ready, |
| }, |
| "sensors": { |
| "google,foo": { |
| "compatible": { |
| "org": "google", |
| "part": "foo", |
| }, |
| "attributes": { |
| "sample_rate": expected_attribute_sample_rate, |
| }, |
| "triggers": { |
| "data_ready": expected_trigger_data_ready, |
| }, |
| "channels": { |
| "bar": expected_sensor_channel_bar, |
| "soap": expected_sensor_channel_soap, |
| "laundry_shirts": ( |
| expected_sensor_channel_laundry_shirts |
| ), |
| "laundry_pants": ( |
| expected_sensor_channel_laundry_pants |
| ), |
| "laundry": expected_sensor_channel_laundry, |
| }, |
| }, |
| }, |
| }, |
| ) |
| |
| 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(), exception_string) |
| 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() |
