• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python
2#
3# Copyright 2016 - 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
17"""Tests for acloud.public.device_driver."""
18
19import uuid
20
21import unittest
22
23from unittest import mock
24
25from acloud.internal.lib import auth
26from acloud.internal.lib import android_build_client
27from acloud.internal.lib import android_compute_client
28from acloud.internal.lib import driver_test_lib
29from acloud.internal.lib import gstorage_client
30from acloud.internal.lib import ssh
31from acloud.public import device_driver
32
33
34def _CreateCfg():
35    """A helper method that creates a mock configuration object."""
36    cfg = mock.MagicMock()
37    cfg.service_account_name = "fake@service.com"
38    cfg.service_account_private_key_path = "/fake/path/to/key"
39    cfg.zone = "fake_zone"
40    cfg.disk_image_name = "fake_image.tar.gz"
41    cfg.disk_image_mime_type = "fake/type"
42    cfg.storage_bucket_name = "fake_bucket"
43    cfg.extra_data_disk_size_gb = 4
44    cfg.precreated_data_image_map = {
45        4: "extradisk-image-4gb",
46        10: "extradisk-image-10gb"
47    }
48    cfg.extra_scopes = None
49    cfg.ssh_private_key_path = ""
50    cfg.ssh_public_key_path = ""
51
52    return cfg
53
54
55class DeviceDriverTest(driver_test_lib.BaseDriverTest):
56    """Test device_driver."""
57
58    def setUp(self):
59        """Set up the test."""
60        super(DeviceDriverTest, self).setUp()
61        self.build_client = mock.MagicMock()
62        self.Patch(android_build_client, "AndroidBuildClient",
63                   return_value=self.build_client)
64        self.storage_client = mock.MagicMock()
65        self.Patch(
66            gstorage_client, "StorageClient", return_value=self.storage_client)
67        self.compute_client = mock.MagicMock()
68        self.Patch(
69            android_compute_client,
70            "AndroidComputeClient",
71            return_value=self.compute_client)
72        self.Patch(auth, "CreateCredentials", return_value=mock.MagicMock())
73        self.fake_avd_spec = mock.MagicMock()
74        self.fake_avd_spec.unlock_screen = False
75        self.fake_avd_spec.client_adb_port = 1234
76
77    def testCreateGCETypeAVD(self):
78        """Test CreateGCETypeAVD."""
79        cfg = _CreateCfg()
80        fake_gs_url = "fake_gs_url"
81        fake_ip = ssh.IP(external="140.1.1.1", internal="10.1.1.1")
82        fake_instance = "fake-instance"
83        fake_image = "fake-image"
84        fake_build_target = "fake_target"
85        fake_build_id = "12345"
86
87        # Mock uuid
88        fake_uuid = mock.MagicMock(hex="1234")
89        self.Patch(uuid, "uuid4", return_value=fake_uuid)
90        fake_gs_object = fake_uuid.hex + "-" + cfg.disk_image_name
91        self.storage_client.GetUrl.return_value = fake_gs_url
92
93        # Mock compute client methods
94        disk_name = "extradisk-image-4gb"
95        self.compute_client.GetInstanceIP.return_value = fake_ip
96        self.compute_client.GenerateImageName.return_value = fake_image
97        self.compute_client.GenerateInstanceName.return_value = fake_instance
98        self.compute_client.GetDataDiskName.return_value = disk_name
99
100        # Verify
101        report = device_driver.CreateGCETypeAVD(
102            cfg, fake_build_target, fake_build_id, avd_spec=self.fake_avd_spec)
103        self.build_client.CopyTo.assert_called_with(
104            fake_build_target, fake_build_id, artifact_name=cfg.disk_image_name,
105            destination_bucket=cfg.storage_bucket_name,
106            destination_path=fake_gs_object)
107        self.compute_client.CreateImage.assert_called_with(
108            image_name=fake_image, source_uri=fake_gs_url)
109        self.compute_client.CreateInstance.assert_called_with(
110            instance=fake_instance,
111            image_name=fake_image,
112            extra_disk_name=disk_name,
113            avd_spec=self.fake_avd_spec,
114            extra_scopes=None)
115        self.compute_client.DeleteImage.assert_called_with(fake_image)
116        self.storage_client.Delete(cfg.storage_bucket_name, fake_gs_object)
117
118        self.assertEqual(
119            report.data,
120            {
121                "devices": [
122                    {
123                        "instance_name": fake_instance,
124                        "ip": fake_ip.external,
125                    },
126                ],
127            }
128        )
129        self.assertEqual(report.command, "create")
130        self.assertEqual(report.status, "SUCCESS")
131
132    # pylint: disable=invalid-name
133    def testCreateGCETypeAVDInternalIP(self):
134        """Test CreateGCETypeAVD with internal IP."""
135        cfg = _CreateCfg()
136        fake_ip = ssh.IP(external="140.1.1.1", internal="10.1.1.1")
137        fake_instance = "fake-instance"
138        fake_build_target = "fake_target"
139        fake_build_id = "12345"
140
141        self.compute_client.GetInstanceIP.return_value = fake_ip
142        self.compute_client.GenerateInstanceName.return_value = fake_instance
143
144        report = device_driver.CreateGCETypeAVD(
145            cfg, fake_build_target, fake_build_id, report_internal_ip=True,
146            avd_spec=self.fake_avd_spec)
147
148        self.assertEqual(
149            report.data,
150            {
151                "devices": [
152                    {
153                        "instance_name": fake_instance,
154                        "ip": fake_ip.internal,
155                    },
156                ],
157            }
158        )
159
160    def testDeleteAndroidVirtualDevices(self):
161        """Test DeleteAndroidVirtualDevices."""
162        cfg = _CreateCfg()
163        instance_names = ["fake-instance-1", "fake-instance-2"]
164        self.compute_client.GetZonesByInstances.return_value = (
165            {cfg.zone: instance_names})
166        self.compute_client.DeleteInstances.return_value = (instance_names, [],
167                                                            [])
168        report = device_driver.DeleteAndroidVirtualDevices(cfg, instance_names)
169        self.compute_client.DeleteInstances.assert_called_once_with(
170            instance_names, cfg.zone)
171        self.assertEqual(report.data, {
172            "deleted": [
173                {
174                    "name": instance_names[0],
175                    "type": "instance",
176                },
177                {
178                    "name": instance_names[1],
179                    "type": "instance",
180                },
181            ],
182        })
183        self.assertEqual(report.command, "delete")
184        self.assertEqual(report.status, "SUCCESS")
185
186
187if __name__ == "__main__":
188    unittest.main()
189