aval/test_device.py at main · torizon/aval · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import unittest
from unittest.mock import MagicMock, patch

with patch("logging_setup.setup_logging", return_value=MagicMock()):
    from device import Device

max_attempts = 10


class TestDeviceGetCurrentBuild(unittest.TestCase):
    def setUp(self):
        patcher_logger = patch("device.logger")
        self.addCleanup(patcher_logger.stop)
        self.logger = patcher_logger.start()

        self.mock_cloud_api = MagicMock()
        self.uuid = "device-uuid-1234"
        self.hardware_id = "component-x"
        self.env_vars = {
            "DEVICE_PASSWORD": "pass",
            "PUBLIC_KEY": "ssh-rsa AAAAB3Nza...",
            "USE_RAC": False,
        }

        with patch.object(
            Device, "_get_network_info", return_value={"localIpV4": "127.0.0.1"}
        ):
            self.device = Device(
                cloud_api=self.mock_cloud_api,
                uuid=self.uuid,
                hardware_id=self.hardware_id,
                env_vars=self.env_vars,
            )

    @patch("time.sleep", return_value=None)
    def test_get_current_build_success(self, _):
        expected_build = "my-build-1.2.3"

        self.mock_cloud_api.get_package_metadata_for_device.return_value = [
            {
                "installedPackages": [
                    {
                        "component": self.hardware_id,
                        "installed": {"packageId": expected_build},
                    }
                ]
            }
        ]

        result = self.device.get_current_build()
        self.assertEqual(result, expected_build)
        self.mock_cloud_api.get_package_metadata_for_device.assert_called_once_with(
            self.uuid
        )

    @patch("time.sleep", return_value=None)
    def test_get_current_build_retry_and_succeed(self, _):
        expected_build = "my-build-1.2.3"

        # First two attempts raise exception, third succeeds
        self.mock_cloud_api.get_package_metadata_for_device.side_effect = [
            Exception("API error"),
            Exception("API error again"),
            [
                {
                    "installedPackages": [
                        {
                            "component": self.hardware_id,
                            "installed": {"packageId": expected_build},
                        }
                    ]
                }
            ],
        ]

        with patch("time.sleep", return_value=None):
            result = self.device.get_current_build()

        self.assertEqual(result, expected_build)
        self.assertEqual(
            self.mock_cloud_api.get_package_metadata_for_device.call_count, 3
        )

    @patch("time.sleep", return_value=None)
    def test_get_current_build_failure_no_packages(self, _):
        self.mock_cloud_api.get_package_metadata_for_device.return_value = [{}]

        with self.assertRaises(Exception) as context:
            self.device.get_current_build()

        expected_message = f"Couldn't parse the current build for {self.device.uuid} after {max_attempts} attempts"
        self.assertIn(expected_message, str(context.exception))
        self.assertEqual(
            self.mock_cloud_api.get_package_metadata_for_device.call_count,
            max_attempts,
        )

    @patch("time.sleep", return_value=None)
    def test_get_current_build_failure_hardware_not_found(self, _):
        self.mock_cloud_api.get_package_metadata_for_device.return_value = [
            {
                "installedPackages": [
                    {
                        "component": "wrong-hardware",
                        "installed": {"packageId": "some-build"},
                    }
                ]
            }
        ]

        with self.assertRaises(Exception) as context:
            self.device.get_current_build()

        expected_message = f"Couldn't parse the current build for {self.device.uuid} after {max_attempts} attempts"
        self.assertIn(expected_message, str(context.exception))
        self.assertEqual(
            self.mock_cloud_api.get_package_metadata_for_device.call_count,
            max_attempts,
        )


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