ardupilot/Tools/autotest/unittest/annotate_params_unittest.py

#!/usr/bin/python3

'''
These are the unit tests for the python script that fetches online ArduPilot
parameter documentation (if not cached) and adds it to the specified file or
to all *.param and *.parm files in the specified directory.

AP_FLAKE8_CLEAN

Author: Amilcar do Carmo Lucas, IAV GmbH
'''

import tempfile
from unittest.mock import patch, mock_open
import os
import unittest
import xml.etree.ElementTree as ET
import requests
import mock
from annotate_params import get_xml_data, remove_prefix, split_into_lines, create_doc_dict, \
                            format_columns, update_parameter_documentation, print_read_only_params, \
                            BASE_URL, PARAM_DEFINITION_XML_FILE


class TestParamDocsUpdate(unittest.TestCase):

    def setUp(self):
        # Create a temporary directory
        self.temp_dir = tempfile.mkdtemp()

        # Create a temporary file
        self.temp_file = tempfile.NamedTemporaryFile(delete=False)

        # Create a dictionary of parameter documentation
        self.doc_dict = {
            "PARAM1": {
                "humanName": "Param 1",
                "documentation": ["Documentation for Param 1"],
                "fields": {"Field1": "Value1", "Field2": "Value2"},
                "values": {"Code1": "Value1", "Code2": "Value2"}
            },
            "PARAM2": {
                "humanName": "Param 2",
                "documentation": ["Documentation for Param 2"],
                "fields": {"Field3": "Value3", "Field4": "Value4"},
                "values": {"Code3": "Value3", "Code4": "Value4"}
            },
            "PARAM_1": {
                "humanName": "Param _ 1",
                "documentation": ["Documentation for Param_1"],
                "fields": {"Field_1": "Value_1", "Field_2": "Value_2"},
                "values": {"Code_1": "Value_1", "Code_2": "Value_2"}
            },
        }

    @patch('builtins.open', new_callable=mock_open, read_data='<root></root>')
    @patch('os.path.isfile')
    def test_get_xml_data_local_file(self, mock_isfile, mock_open):
        # Mock the isfile function to return True
        mock_isfile.return_value = True

        # Call the function with a local file
        result = get_xml_data("/path/to/local/file/", ".", "test.xml")

        # Check the result
        self.assertIsInstance(result, ET.Element)

        # Assert that the file was opened correctly
        mock_open.assert_called_once_with('./test.xml', 'r', encoding='utf-8')

    @patch('requests.get')
    def test_get_xml_data_remote_file(self, mock_get):
        # Mock the response
        mock_get.return_value.status_code = 200
        mock_get.return_value.text = "<root></root>"

        # Remove the test.xml file if it exists
        try:
            os.remove("test.xml")
        except FileNotFoundError:
            pass

        # Call the function with a remote file
        result = get_xml_data("http://example.com/", ".", "test.xml")

        # Check the result
        self.assertIsInstance(result, ET.Element)

        # Assert that the requests.get function was called once
        mock_get.assert_called_once_with("http://example.com/test.xml", timeout=5)

    @patch('os.path.isfile')
    def test_get_xml_data_script_dir_file(self, mock_isfile):
        # Mock the isfile function to return False for the current directory and True for the script directory
        def side_effect(filename):
            return True
        mock_isfile.side_effect = side_effect

        # Mock the open function to return a dummy XML string
        mock_open = mock.mock_open(read_data='<root></root>')
        with patch('builtins.open', mock_open):
            # Call the function with a filename that exists in the script directory
            result = get_xml_data(BASE_URL, ".", PARAM_DEFINITION_XML_FILE)

        # Check the result
        self.assertIsInstance(result, ET.Element)

        # Assert that the file was opened correctly
        mock_open.assert_called_once_with(os.path.join('.', PARAM_DEFINITION_XML_FILE), 'r', encoding='utf-8')

    def test_get_xml_data_no_requests_package(self):
        # Temporarily remove the requests module
        with patch.dict('sys.modules', {'requests': None}):

            # Remove the test.xml file if it exists
            try:
                os.remove("test.xml")
            except FileNotFoundError:
                pass

            # Call the function with a remote file
            with self.assertRaises(SystemExit):
                get_xml_data("http://example.com/", ".", "test.xml")

    @patch('requests.get')
    def test_get_xml_data_request_failure(self, mock_get):
        # Mock the response
        mock_get.side_effect = requests.exceptions.RequestException

        # Remove the test.xml file if it exists
        try:
            os.remove("test.xml")
        except FileNotFoundError:
            pass

        # Call the function with a remote file
        with self.assertRaises(SystemExit):
            get_xml_data("http://example.com/", ".", "test.xml")

    @patch('requests.get')
    def test_get_xml_data_valid_xml(self, mock_get):
        # Mock the response
        mock_get.return_value.status_code = 200
        mock_get.return_value.text = "<root></root>"

        # Call the function with a remote file
        result = get_xml_data("http://example.com/", ".", "test.xml")

        # Check the result
        self.assertIsInstance(result, ET.Element)

    @patch('requests.get')
    def test_get_xml_data_invalid_xml(self, mock_get):
        # Mock the response
        mock_get.return_value.status_code = 200
        mock_get.return_value.text = "<root><invalid></root>"

        # Remove the test.xml file if it exists
        try:
            os.remove("test.xml")
        except FileNotFoundError:
            pass

        # Call the function with a remote file
        with self.assertRaises(ET.ParseError):
            get_xml_data("http://example.com/", ".", "test.xml")

    @patch('requests.get')
    @patch('os.path.isfile')
    def test_get_xml_data_missing_file(self, mock_isfile, mock_get):
        # Mock the isfile function to return False
        mock_isfile.return_value = False
        # Mock the requests.get call to raise FileNotFoundError
        mock_get.side_effect = FileNotFoundError

        # Remove the test.xml file if it exists
        try:
            os.remove("test.xml")
        except FileNotFoundError:
            pass

        # Call the function with a local file
        with self.assertRaises(FileNotFoundError):
            get_xml_data("/path/to/local/file/", ".", "test.xml")

    @patch('requests.get')
    def test_get_xml_data_network_issue(self, mock_get):
        # Mock the response
        mock_get.side_effect = requests.exceptions.ConnectionError

        # Call the function with a remote file
        with self.assertRaises(SystemExit):
            get_xml_data("http://example.com/", ".", "test.xml")

    def test_remove_prefix(self):
        # Test case 1: Normal operation
        self.assertEqual(remove_prefix("prefix_test", "prefix_"), "test")

        # Test case 2: Prefix not present
        self.assertEqual(remove_prefix("test", "prefix_"), "test")

        # Test case 3: Empty string
        self.assertEqual(remove_prefix("", "prefix_"), "")

    def test_split_into_lines(self):
        # Test case 1: Normal operation
        string_to_split = "This is a test string. It should be split into several lines."
        maximum_line_length = 12
        expected_output = ["This is a", "test string.", "It should be", "split into", "several",  "lines."]
        self.assertEqual(split_into_lines(string_to_split, maximum_line_length), expected_output)

        # Test case 2: String shorter than maximum line length
        string_to_split = "Short"
        maximum_line_length = 10
        expected_output = ["Short"]
        self.assertEqual(split_into_lines(string_to_split, maximum_line_length), expected_output)

        # Test case 3: Empty string
        string_to_split = ""
        maximum_line_length = 10
        expected_output = []
        self.assertEqual(split_into_lines(string_to_split, maximum_line_length), expected_output)

    def test_create_doc_dict(self):
        # Mock XML data
        xml_data = '''
        <root>
            <param name="PARAM1" humanName="Param 1" documentation="Documentation for Param 1">
                <field name="Field1">Value1</field>
                <field name="Field2">Value2</field>
                <values>
                    <value code="Code1">Value1</value>
                    <value code="Code2">Value2</value>
                </values>
            </param>
            <param name="PARAM2" humanName="Param 2" documentation="Documentation for Param 2">
                <field name="Units">m/s</field>
                <field name="UnitText">meters per second</field>
                <values>
                    <value code="Code3">Value3</value>
                    <value code="Code4">Value4</value>
                </values>
            </param>
        </root>
        '''
        root = ET.fromstring(xml_data)

        # Expected output
        expected_output = {
            "PARAM1": {
                "humanName": "Param 1",
                "documentation": ["Documentation for Param 1"],
                "fields": {"Field1": "Value1", "Field2": "Value2"},
                "values": {"Code1": "Value1", "Code2": "Value2"}
            },
            "PARAM2": {
                "humanName": "Param 2",
                "documentation": ["Documentation for Param 2"],
                "fields": {"Units": "m/s (meters per second)"},
                "values": {"Code3": "Value3", "Code4": "Value4"}
            }
        }

        # Call the function with the mock XML data
        result = create_doc_dict(root, "VehicleType")

        # Check the result
        self.assertEqual(result, expected_output)

    def test_format_columns(self):
        # Define the input
        values = {
            "Key1": "Value1",
            "Key2": "Value2",
            "Key3": "Value3",
            "Key4": "Value4",
            "Key5": "Value5",
            "Key6": "Value6",
            "Key7": "Value7",
            "Key8": "Value8",
            "Key9": "Value9",
            "Key10": "Value10",
            "Key11": "Value11",
            "Key12": "Value12",
        }

        # Define the expected output
        expected_output = [
            'Key1: Value1                                         Key7: Value7',
            'Key2: Value2                                         Key8: Value8',
            'Key3: Value3                                         Key9: Value9',
            'Key4: Value4                                         Key10: Value10',
            'Key5: Value5                                         Key11: Value11',
            'Key6: Value6                                         Key12: Value12',
        ]

        # Call the function with the input
        result = format_columns(values)

        # Check the result
        self.assertEqual(result, expected_output)

        self.assertEqual(format_columns({}), [])

    def test_update_parameter_documentation(self):
        # Write some initial content to the temporary file
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("PARAM1 100\n")

        # Call the function with the temporary file
        update_parameter_documentation(self.doc_dict, self.temp_file.name)

        # Read the updated content from the temporary file
        with open(self.temp_file.name, "r", encoding="utf-8") as file:
            updated_content = file.read()

        # Check if the file has been updated correctly
        self.assertIn("Param 1", updated_content)
        self.assertIn("Documentation for Param 1", updated_content)
        self.assertIn("Field1: Value1", updated_content)
        self.assertIn("Field2: Value2", updated_content)
        self.assertIn("Code1: Value1", updated_content)
        self.assertIn("Code2: Value2", updated_content)

    def test_update_parameter_documentation_sorting_none(self):
        # Write some initial content to the temporary file
        # With stray leading and trailing whitespaces
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("PARAM2 100\n PARAM_1 100 \nPARAM3 3\nPARAM4 4\nPARAM5 5\nPARAM1 100\n")

        # Call the function with the temporary file
        update_parameter_documentation(self.doc_dict, self.temp_file.name)

        # Read the updated content from the temporary file
        with open(self.temp_file.name, "r", encoding="utf-8") as file:
            updated_content = file.read()

        expected_content = '''# Param 2
# Documentation for Param 2
# Field3: Value3
# Field4: Value4
# Code3: Value3
# Code4: Value4
PARAM2 100

# Param _ 1
# Documentation for Param_1
# Field_1: Value_1
# Field_2: Value_2
# Code_1: Value_1
# Code_2: Value_2
PARAM_1 100
PARAM3 3
PARAM4 4
PARAM5 5

# Param 1
# Documentation for Param 1
# Field1: Value1
# Field2: Value2
# Code1: Value1
# Code2: Value2
PARAM1 100
'''
        self.assertEqual(updated_content, expected_content)

    def test_update_parameter_documentation_sorting_missionplanner(self):
        # Write some initial content to the temporary file
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("PARAM2 100 # ignore, me\nPARAM_1\t100\nPARAM1,100\n")

        # Call the function with the temporary file
        update_parameter_documentation(self.doc_dict, self.temp_file.name, "missionplanner")

        # Read the updated content from the temporary file
        with open(self.temp_file.name, "r", encoding="utf-8") as file:
            updated_content = file.read()

        expected_content = '''# Param _ 1
# Documentation for Param_1
# Field_1: Value_1
# Field_2: Value_2
# Code_1: Value_1
# Code_2: Value_2
PARAM_1\t100

# Param 1
# Documentation for Param 1
# Field1: Value1
# Field2: Value2
# Code1: Value1
# Code2: Value2
PARAM1,100

# Param 2
# Documentation for Param 2
# Field3: Value3
# Field4: Value4
# Code3: Value3
# Code4: Value4
PARAM2 100 # ignore, me
'''
        self.assertEqual(updated_content, expected_content)

    def test_update_parameter_documentation_sorting_mavproxy(self):
        # Write some initial content to the temporary file
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("PARAM2 100\nPARAM_1\t100\nPARAM1,100\n")

        # Call the function with the temporary file
        update_parameter_documentation(self.doc_dict, self.temp_file.name, "mavproxy")

        # Read the updated content from the temporary file
        with open(self.temp_file.name, "r", encoding="utf-8") as file:
            updated_content = file.read()

        expected_content = '''# Param 1
# Documentation for Param 1
# Field1: Value1
# Field2: Value2
# Code1: Value1
# Code2: Value2
PARAM1,100

# Param 2
# Documentation for Param 2
# Field3: Value3
# Field4: Value4
# Code3: Value3
# Code4: Value4
PARAM2 100

# Param _ 1
# Documentation for Param_1
# Field_1: Value_1
# Field_2: Value_2
# Code_1: Value_1
# Code_2: Value_2
PARAM_1\t100
'''
        self.assertEqual(updated_content, expected_content)

    def test_update_parameter_documentation_invalid_line_format(self):
        # Write some initial content to the temporary file with an invalid line format
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("%INVALID_LINE_FORMAT\n")

        # Call the function with the temporary file
        with self.assertRaises(SystemExit) as cm:
            update_parameter_documentation(self.doc_dict, self.temp_file.name)

        # Check if the SystemExit exception contains the expected message
        self.assertEqual(cm.exception.code, "Invalid line in input file")

    @patch('logging.Logger.info')
    def test_print_read_only_params(self, mock_info):
        # Mock XML data
        xml_data = '''
        <root>
            <param name="PARAM1" humanName="Param 1" documentation="Documentation for Param 1">
                <field name="ReadOnly">True</field>
                <field name="Field1">Value1</field>
                <field name="Field2">Value2</field>
                <values>
                    <value code="Code1">Value1</value>
                    <value code="Code2">Value2</value>
                </values>
            </param>
            <param name="PARAM2" humanName="Param 2" documentation="Documentation for Param 2">
                <field name="Field3">Value3</field>
                <field name="Field4">Value4</field>
                <values>
                    <value code="Code3">Value3</value>
                    <value code="Code4">Value4</value>
                </values>
            </param>
        </root>
        '''
        root = ET.fromstring(xml_data)
        doc_dict = create_doc_dict(root, "VehicleType")

        # Call the function with the mock XML data
        print_read_only_params(doc_dict)

        # Check if the parameter name was logged
        mock_info.assert_has_calls([mock.call('ReadOnly parameters:'), mock.call('PARAM1')])

    def test_update_parameter_documentation_invalid_target(self):
        # Call the function with an invalid target
        with self.assertRaises(ValueError):
            update_parameter_documentation(self.doc_dict, "invalid_target")

    def test_invalid_parameter_name(self):
        # Write some initial content to the temporary file
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("INVALID_$PARAM 100\n")

        # Call the function with the temporary file
        with self.assertRaises(SystemExit):
            update_parameter_documentation(self.doc_dict, self.temp_file.name)

    def test_update_parameter_documentation_too_long_parameter_name(self):
        # Write some initial content to the temporary file
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("TOO_LONG_PARAMETER_NAME 100\n")

        # Call the function with the temporary file
        with self.assertRaises(SystemExit):
            update_parameter_documentation(self.doc_dict, self.temp_file.name)

    @patch('logging.Logger.warning')
    def test_missing_parameter_documentation(self, mock_warning):
        # Write some initial content to the temporary file
        with open(self.temp_file.name, "w", encoding="utf-8") as file:
            file.write("MISSING_DOC_PARA 100\n")

        # Call the function with the temporary file
        update_parameter_documentation(self.doc_dict, self.temp_file.name)

        # Check if the warnings were logged
        mock_warning.assert_has_calls([
            mock.call('Read file %s with %d parameters, but only %s of which got documented', self.temp_file.name, 1, 0),
            mock.call('No documentation found for: %s', 'MISSING_DOC_PARA')
        ])

    def test_empty_parameter_file(self):
        # Call the function with the temporary file
        update_parameter_documentation(self.doc_dict, self.temp_file.name)

        # Read the updated content from the temporary file
        with open(self.temp_file.name, "r", encoding="utf-8") as file:
            updated_content = file.read()

        # Check if the file is still empty
        self.assertEqual(updated_content, "")


if __name__ == '__main__':
    unittest.main()