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
48class OtaUpdaterTests(unittest.TestCase):
49    """Tests the methods in the ota_updater module."""
50
51    def test_initialize(self):
52        user_params = {'a': 1, 'b': 2, 'c': 3}
53        android_devices = ['x', 'y', 'z']
54        with mock.patch('acts.libs.ota.ota_runners.ota_runner_factory.'
55                        'create_from_configs') as fn:
56            ota_updater.initialize(user_params, android_devices)
57            for i in range(len(android_devices)):
58                fn.assert_has_call(mock.call(user_params, android_devices[i]))
59        self.assertSetEqual(
60            set(android_devices), set(ota_updater.ota_runners.keys()))
61
62    def test_check_initialization_is_initialized(self):
63        device = MockAndroidDevice('serial')
64        ota_updater.ota_runners = {
65            device: ota_runner.OtaRunner('tool', device)
66        }
67        try:
68            ota_updater._check_initialization(device)
69        except ota_runner.OtaError:
70            self.fail('_check_initialization raised for initialized runner!')
71
72    def test_check_initialization_is_not_initialized(self):
73        device = MockAndroidDevice('serial')
74        ota_updater.ota_runners = {}
75        with self.assertRaises(KeyError):
76            ota_updater._check_initialization(device)
77
78    def test_update_do_not_ignore_failures_and_failures_occur(self):
79        device = MockAndroidDevice('serial')
80        runner = MockOtaRunner()
81        runner.set_failure(True)
82        ota_updater.ota_runners = {device: runner}
83        with self.assertRaises(ota_runner.OtaError):
84            ota_updater.update(device)
85
86    def test_update_ignore_failures_and_failures_occur(self):
87        device = MockAndroidDevice('serial')
88        runner = MockOtaRunner()
89        runner.set_failure(True)
90        ota_updater.ota_runners = {device: runner}
91        try:
92            ota_updater.update(device, ignore_update_errors=True)
93        except ota_runner.OtaError:
94            self.fail('OtaError was raised when errors are to be ignored!')
95
96    def test_can_update(self):
97        device = MockAndroidDevice('serial')
98        runner = MockOtaRunner()
99        ota_updater.ota_runners = {device: runner}
100        self.assertEqual(ota_updater.can_update(device), 'CAN_UPDATE_CALLED')
101