pw_sensor/py/validator_test.py

# 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()