1#!/usr/bin/env python3
2#
3#   Copyright 2017 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import mock
18import unittest
19from acts.libs.ota import ota_updater
20from acts.libs.ota.ota_runners import ota_runner
21
22
23class MockAndroidDevice(object):
24    def __init__(self, serial):
25        self.serial = serial
26        self.log = mock.Mock()
27        self.take_bug_report = mock.MagicMock()
28
29
30class MockOtaRunner(object):
31    def __init__(self):
32        self.call_count = 0
33        self.should_fail = False
34        self.can_update_value = 'CAN_UPDATE_CALLED'
35
36    def set_failure(self, should_fail=True):
37        self.should_fail = should_fail
38
39    def update(self):
40        self.call_count += 1
41        if self.should_fail:
42            raise ota_runner.OtaError
43
44    def can_update(self):
45        return self.can_update_value
46
47    def validate_update(self):
48        pass
49
50
51class OtaUpdaterTests(unittest.TestCase):
52    """Tests the methods in the ota_updater module."""
53
54    def test_initialize(self):
55        user_params = {'a': 1, 'b': 2, 'c': 3}
56        android_devices = ['x', 'y', 'z']
57        with mock.patch('acts.libs.ota.ota_runners.ota_runner_factory.'
58                        'create_from_configs') as fn:
59            ota_updater.initialize(user_params, android_devices)
60            for i in range(len(android_devices)):
61                fn.assert_any_call(user_params, android_devices[i])
62            self.assertSetEqual(
63                set(android_devices), set(ota_updater.ota_runners.keys()))
64
65    def test_check_initialization_is_initialized(self):
66        device = MockAndroidDevice('serial')
67        ota_updater.ota_runners = {
68            device: ota_runner.OtaRunner('tool', device)
69        }
70        try:
71            ota_updater._check_initialization(device)
72        except ota_runner.OtaError:
73            self.fail(
74                '_check_initialization raised for initialized runner!')
75
76    def test_check_initialization_is_not_initialized(self):
77        device = MockAndroidDevice('serial')
78        ota_updater.ota_runners = {}
79        with self.assertRaises(KeyError):
80            ota_updater._check_initialization(device)
81
82    def test_update_do_not_ignore_failures_and_failures_occur(self):
83        device = MockAndroidDevice('serial')
84        runner = MockOtaRunner()
85        runner.set_failure(True)
86        ota_updater.ota_runners = {device: runner}
87        with self.assertRaises(ota_runner.OtaError):
88            ota_updater.update(device)
89
90    def test_update_ignore_failures_and_failures_occur(self):
91        device = MockAndroidDevice('serial')
92        runner = MockOtaRunner()
93        runner.set_failure(True)
94        ota_updater.ota_runners = {device: runner}
95        try:
96            ota_updater.update(device, ignore_update_errors=True)
97        except ota_runner.OtaError:
98            self.fail('OtaError was raised when errors are to be ignored!')
99
100    def test_can_update(self):
101        device = MockAndroidDevice('serial')
102        runner = MockOtaRunner()
103        ota_updater.ota_runners = {device: runner}
104        self.assertEqual(ota_updater.can_update(device),
105                         'CAN_UPDATE_CALLED')
106
107
108if __name__ == '__main__':
109    unittest.main()
110