1# Copyright 2018 - The Android Open Source Project
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Tests for avd_spec."""
15
16import glob
17import os
18import subprocess
19import unittest
20import mock
21
22from acloud import errors
23from acloud.create import avd_spec
24from acloud.internal import constants
25from acloud.internal.lib import android_build_client
26from acloud.internal.lib import auth
27from acloud.internal.lib import driver_test_lib
28from acloud.internal.lib import utils
29from acloud.list import list as list_instances
30
31
32# pylint: disable=invalid-name,protected-access
33class AvdSpecTest(driver_test_lib.BaseDriverTest):
34    """Test avd_spec methods."""
35
36    def setUp(self):
37        """Initialize new avd_spec.AVDSpec."""
38        super(AvdSpecTest, self).setUp()
39        self.args = mock.MagicMock()
40        self.args.flavor = ""
41        self.args.local_image = ""
42        self.args.config_file = ""
43        self.args.build_target = "fake_build_target"
44        self.args.adb_port = None
45        self.Patch(list_instances, "ChooseOneRemoteInstance", return_value=mock.MagicMock())
46        self.Patch(list_instances, "GetInstancesFromInstanceNames", return_value=mock.MagicMock())
47        self.AvdSpec = avd_spec.AVDSpec(self.args)
48
49    # pylint: disable=protected-access
50    def testProcessLocalImageArgs(self):
51        """Test process args.local_image."""
52        self.Patch(glob, "glob", return_value=["fake.img"])
53        expected_image_artifact = "/path/cf_x86_phone-img-eng.user.zip"
54        expected_image_dir = "/path-to-image-dir"
55
56        # Specified --local-image to a local zipped image file
57        self.Patch(os.path, "isfile", return_value=True)
58        self.args.local_image = "/path/cf_x86_phone-img-eng.user.zip"
59        self.AvdSpec._avd_type = constants.TYPE_CF
60        self.AvdSpec._instance_type = constants.INSTANCE_TYPE_REMOTE
61        self.AvdSpec._ProcessLocalImageArgs(self.args)
62        self.assertEqual(self.AvdSpec._local_image_artifact,
63                         expected_image_artifact)
64
65        # Specified --local-image to a dir contains images
66        self.Patch(utils, "GetBuildEnvironmentVariable",
67                   return_value="test_cf_x86")
68        self.Patch(os.path, "isfile", return_value=False)
69        self.args.local_image = "/path-to-image-dir"
70        self.AvdSpec._avd_type = constants.TYPE_CF
71        self.AvdSpec._instance_type = constants.INSTANCE_TYPE_REMOTE
72        self.AvdSpec._ProcessLocalImageArgs(self.args)
73        self.assertEqual(self.AvdSpec._local_image_dir, expected_image_dir)
74
75        # Specified local_image without arg
76        self.args.local_image = None
77        self.Patch(utils, "GetBuildEnvironmentVariable",
78                   side_effect=["cf_x86_auto", "test_environ", "test_environ"])
79        self.AvdSpec._ProcessLocalImageArgs(self.args)
80        self.assertEqual(self.AvdSpec._local_image_dir, "test_environ")
81        self.assertEqual(self.AvdSpec.local_image_artifact, expected_image_artifact)
82
83        # Specified --avd-type=goldfish --local-image with a dir
84        self.Patch(utils, "GetBuildEnvironmentVariable",
85                   return_value="test_environ")
86        self.Patch(os.path, "isdir", return_value=True)
87        self.args.local_image = "/path-to-image-dir"
88        self.AvdSpec._avd_type = constants.TYPE_GF
89        self.AvdSpec._instance_type = constants.INSTANCE_TYPE_LOCAL
90        self.AvdSpec._ProcessLocalImageArgs(self.args)
91        self.assertEqual(self.AvdSpec._local_image_dir, expected_image_dir)
92
93        # Specified --avd-type=goldfish --local_image without arg
94        self.Patch(utils, "GetBuildEnvironmentVariable",
95                   return_value="test_environ")
96        self.Patch(os.path, "isdir", return_value=True)
97        self.args.local_image = None
98        self.AvdSpec._avd_type = constants.TYPE_GF
99        self.AvdSpec._instance_type = constants.INSTANCE_TYPE_LOCAL
100        self.AvdSpec._ProcessLocalImageArgs(self.args)
101        self.assertEqual(self.AvdSpec._local_image_dir, "test_environ")
102
103    def testProcessImageArgs(self):
104        """Test process image source."""
105        self.Patch(glob, "glob", return_value=["fake.img"])
106        # No specified local_image, image source is from remote
107        self.args.local_image = ""
108        self.AvdSpec._ProcessImageArgs(self.args)
109        self.assertEqual(self.AvdSpec._image_source, constants.IMAGE_SRC_REMOTE)
110        self.assertEqual(self.AvdSpec._local_image_dir, None)
111
112        # Specified local_image with an arg for cf type
113        self.Patch(os.path, "isfile", return_value=True)
114        self.args.local_image = "/test_path/cf_x86_phone-img-eng.user.zip"
115        self.AvdSpec._avd_type = constants.TYPE_CF
116        self.AvdSpec._instance_type = constants.INSTANCE_TYPE_REMOTE
117        self.AvdSpec._ProcessImageArgs(self.args)
118        self.assertEqual(self.AvdSpec._image_source, constants.IMAGE_SRC_LOCAL)
119        self.assertEqual(self.AvdSpec._local_image_artifact,
120                         "/test_path/cf_x86_phone-img-eng.user.zip")
121
122        # Specified local_image with an arg for gce type
123        self.Patch(os.path, "isfile", return_value=False)
124        self.Patch(os.path, "exists", return_value=True)
125        self.args.local_image = "/test_path_to_dir/"
126        self.AvdSpec._avd_type = constants.TYPE_GCE
127        self.AvdSpec._ProcessImageArgs(self.args)
128        self.assertEqual(self.AvdSpec._image_source, constants.IMAGE_SRC_LOCAL)
129        self.assertEqual(self.AvdSpec._local_image_artifact,
130                         "/test_path_to_dir/avd-system.tar.gz")
131
132    @mock.patch.object(avd_spec.AVDSpec, "_GetGitRemote")
133    def testGetBranchFromRepo(self, mock_gitremote):
134        """Test get branch name from repo info."""
135        # Check aosp repo gets proper branch prefix.
136        fake_subprocess = mock.MagicMock()
137        fake_subprocess.stdout = mock.MagicMock()
138        fake_subprocess.stdout.readline = mock.MagicMock(return_value='')
139        fake_subprocess.poll = mock.MagicMock(return_value=0)
140        fake_subprocess.returncode = 0
141        return_value = b"Manifest branch: master"
142        fake_subprocess.communicate = mock.MagicMock(return_value=(return_value, ''))
143        self.Patch(subprocess, "Popen", return_value=fake_subprocess)
144
145        mock_gitremote.return_value = "aosp"
146        self.assertEqual(self.AvdSpec._GetBranchFromRepo(), "aosp-master")
147
148        # Check default repo gets default branch prefix.
149        mock_gitremote.return_value = ""
150        return_value = b"Manifest branch: master"
151        fake_subprocess.communicate = mock.MagicMock(return_value=(return_value, ''))
152        self.Patch(subprocess, "Popen", return_value=fake_subprocess)
153        self.assertEqual(self.AvdSpec._GetBranchFromRepo(), "git_master")
154
155        # Can't get branch from repo info, set it as default branch.
156        return_value = b"Manifest branch:"
157        fake_subprocess.communicate = mock.MagicMock(return_value=(return_value, ''))
158        self.Patch(subprocess, "Popen", return_value=fake_subprocess)
159        self.assertEqual(self.AvdSpec._GetBranchFromRepo(), "aosp-master")
160
161    def testGetBuildBranch(self):
162        """Test GetBuildBranch function"""
163        # Test infer branch from build_id and build_target.
164        build_client = mock.MagicMock()
165        build_id = "fake_build_id"
166        build_target = "fake_build_target"
167        expected_branch = "fake_build_branch"
168        self.Patch(android_build_client, "AndroidBuildClient",
169                   return_value=build_client)
170        self.Patch(auth, "CreateCredentials", return_value=mock.MagicMock())
171        self.Patch(build_client, "GetBranch", return_value=expected_branch)
172        self.assertEqual(self.AvdSpec._GetBuildBranch(build_id, build_target),
173                         expected_branch)
174        # Infer branch from "repo info" when build_id and build_target is None.
175        self.Patch(self.AvdSpec, "_GetBranchFromRepo", return_value="repo_branch")
176        build_id = None
177        build_target = None
178        expected_branch = "repo_branch"
179        self.assertEqual(self.AvdSpec._GetBuildBranch(build_id, build_target),
180                         expected_branch)
181
182    # pylint: disable=protected-access
183    def testGetBuildTarget(self):
184        """Test get build target name."""
185        self.AvdSpec._remote_image[constants.BUILD_BRANCH] = "git_branch"
186        self.AvdSpec._flavor = constants.FLAVOR_IOT
187        self.args.avd_type = constants.TYPE_GCE
188        self.assertEqual(
189            self.AvdSpec._GetBuildTarget(self.args),
190            "gce_x86_iot-userdebug")
191
192        self.AvdSpec._remote_image[constants.BUILD_BRANCH] = "aosp-master"
193        self.AvdSpec._flavor = constants.FLAVOR_PHONE
194        self.args.avd_type = constants.TYPE_CF
195        self.assertEqual(
196            self.AvdSpec._GetBuildTarget(self.args),
197            "aosp_cf_x86_phone-userdebug")
198
199        self.AvdSpec._remote_image[constants.BUILD_BRANCH] = "git_branch"
200        self.AvdSpec._flavor = constants.FLAVOR_PHONE
201        self.args.avd_type = constants.TYPE_CF
202        self.assertEqual(
203            self.AvdSpec._GetBuildTarget(self.args),
204            "cf_x86_phone-userdebug")
205
206    # pylint: disable=protected-access
207    def testProcessHWPropertyWithInvalidArgs(self):
208        """Test _ProcessHWPropertyArgs with invalid args."""
209        # Checking wrong resolution.
210        args = mock.MagicMock()
211        args.hw_property = "cpu:3,resolution:1280"
212        args.reuse_instance_name = None
213        with self.assertRaises(errors.InvalidHWPropertyError):
214            self.AvdSpec._ProcessHWPropertyArgs(args)
215
216        # Checking property should be int.
217        args = mock.MagicMock()
218        args.hw_property = "cpu:3,dpi:fake"
219        with self.assertRaises(errors.InvalidHWPropertyError):
220            self.AvdSpec._ProcessHWPropertyArgs(args)
221
222        # Checking disk property should be with 'g' suffix.
223        args = mock.MagicMock()
224        args.hw_property = "cpu:3,disk:2"
225        with self.assertRaises(errors.InvalidHWPropertyError):
226            self.AvdSpec._ProcessHWPropertyArgs(args)
227
228        # Checking memory property should be with 'g' suffix.
229        args = mock.MagicMock()
230        args.hw_property = "cpu:3,memory:2"
231        with self.assertRaises(errors.InvalidHWPropertyError):
232            self.AvdSpec._ProcessHWPropertyArgs(args)
233
234    # pylint: disable=protected-access
235    @mock.patch.object(utils, "PrintColorString")
236    def testCheckCFBuildTarget(self, print_warning):
237        """Test _CheckCFBuildTarget."""
238        # patch correct env variable.
239        self.Patch(utils, "GetBuildEnvironmentVariable",
240                   return_value="cf_x86_phone-userdebug")
241        self.AvdSpec._CheckCFBuildTarget(constants.INSTANCE_TYPE_REMOTE)
242        self.AvdSpec._CheckCFBuildTarget(constants.INSTANCE_TYPE_LOCAL)
243
244        self.Patch(utils, "GetBuildEnvironmentVariable",
245                   return_value="aosp_cf_arm64_auto-userdebug")
246        self.AvdSpec._CheckCFBuildTarget(constants.INSTANCE_TYPE_HOST)
247        # patch wrong env variable.
248        self.Patch(utils, "GetBuildEnvironmentVariable",
249                   return_value="test_environ")
250        self.AvdSpec._CheckCFBuildTarget(constants.INSTANCE_TYPE_REMOTE)
251
252        print_warning.assert_called_once()
253
254    # pylint: disable=protected-access
255    def testParseHWPropertyStr(self):
256        """Test _ParseHWPropertyStr."""
257        expected_dict = {"cpu": "2", "x_res": "1080", "y_res": "1920",
258                         "dpi": "240", "memory": "4096", "disk": "4096"}
259        args_str = "cpu:2,resolution:1080x1920,dpi:240,memory:4g,disk:4g"
260        result_dict = self.AvdSpec._ParseHWPropertyStr(args_str)
261        self.assertTrue(expected_dict == result_dict)
262
263        expected_dict = {"cpu": "2", "x_res": "1080", "y_res": "1920",
264                         "dpi": "240", "memory": "512", "disk": "4096"}
265        args_str = "cpu:2,resolution:1080x1920,dpi:240,memory:512m,disk:4g"
266        result_dict = self.AvdSpec._ParseHWPropertyStr(args_str)
267        self.assertTrue(expected_dict == result_dict)
268
269    def testGetFlavorFromBuildTargetString(self):
270        """Test _GetFlavorFromLocalImage."""
271        img_path = "/fack_path/cf_x86_tv-img-eng.user.zip"
272        self.assertEqual(self.AvdSpec._GetFlavorFromString(img_path),
273                         "tv")
274
275        build_target_str = "aosp_cf_x86_auto"
276        self.assertEqual(self.AvdSpec._GetFlavorFromString(
277            build_target_str), "auto")
278
279        # Flavor is not supported.
280        img_path = "/fack_path/cf_x86_error-img-eng.user.zip"
281        self.assertEqual(self.AvdSpec._GetFlavorFromString(img_path),
282                         None)
283
284    # pylint: disable=protected-access
285    def testProcessRemoteBuildArgs(self):
286        """Test _ProcessRemoteBuildArgs."""
287        self.args.branch = "git_master"
288        self.args.build_id = "1234"
289
290        # Verify auto-assigned avd_type if build_targe contains "_gce_".
291        self.args.build_target = "aosp_gce_x86_phone-userdebug"
292        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
293        self.assertTrue(self.AvdSpec.avd_type == "gce")
294
295        # Verify auto-assigned avd_type if build_targe contains "gce_".
296        self.args.build_target = "gce_x86_phone-userdebug"
297        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
298        self.assertTrue(self.AvdSpec.avd_type == "gce")
299
300        # Verify auto-assigned avd_type if build_targe contains "_cf_".
301        self.args.build_target = "aosp_cf_x86_phone-userdebug"
302        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
303        self.assertTrue(self.AvdSpec.avd_type == "cuttlefish")
304
305        # Verify auto-assigned avd_type if build_targe contains "cf_".
306        self.args.build_target = "cf_x86_phone-userdebug"
307        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
308        self.assertTrue(self.AvdSpec.avd_type == "cuttlefish")
309
310        # Verify auto-assigned avd_type if build_targe contains "sdk_".
311        self.args.build_target = "sdk_phone_armv7-sdk"
312        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
313        self.assertTrue(self.AvdSpec.avd_type == "goldfish")
314
315        # Verify auto-assigned avd_type if build_targe contains "_sdk_".
316        self.args.build_target = "aosp_sdk_phone_armv7-sdk"
317        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
318        self.assertTrue(self.AvdSpec.avd_type == "goldfish")
319
320        # Verify auto-assigned avd_type if no match, default as cuttlefish.
321        self.args.build_target = "mini_emulator_arm64-userdebug"
322        self.args.avd_type = "cuttlefish"
323        # reset args.avd_type default value as cuttlefish.
324        self.AvdSpec = avd_spec.AVDSpec(self.args)
325        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
326        self.assertTrue(self.AvdSpec.avd_type == "cuttlefish")
327
328        self.args.cheeps_betty_image = 'abcdefg'
329        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
330        self.assertEqual(
331            self.AvdSpec.remote_image[constants.CHEEPS_BETTY_IMAGE],
332            self.args.cheeps_betty_image)
333        self.args.cheeps_betty_image = None
334        self.AvdSpec._ProcessRemoteBuildArgs(self.args)
335        self.assertEqual(
336            self.AvdSpec.remote_image[constants.CHEEPS_BETTY_IMAGE],
337            self.args.cheeps_betty_image)
338
339
340    def testEscapeAnsi(self):
341        """Test EscapeAnsi."""
342        test_string = "\033[1;32;40m Manifest branch:"
343        expected_result = " Manifest branch:"
344        self.assertEqual(avd_spec.EscapeAnsi(test_string), expected_result)
345
346    def testGetGceLocalImagePath(self):
347        """Test get gce local image path."""
348        self.Patch(os.path, "isfile", return_value=True)
349        # Verify when specify --local-image ~/XXX.tar.gz.
350        fake_image_path = "~/gce_local_image_dir/gce_image.tar.gz"
351        self.Patch(os.path, "exists", return_value=True)
352        self.assertEqual(self.AvdSpec._GetGceLocalImagePath(fake_image_path),
353                         "~/gce_local_image_dir/gce_image.tar.gz")
354
355        # Verify when specify --local-image ~/XXX.img.
356        fake_image_path = "~/gce_local_image_dir/gce_image.img"
357        self.assertEqual(self.AvdSpec._GetGceLocalImagePath(fake_image_path),
358                         "~/gce_local_image_dir/gce_image.img")
359
360        # Verify if exist argument --local-image as a directory.
361        self.Patch(os.path, "isfile", return_value=False)
362        self.Patch(os.path, "exists", return_value=True)
363        fake_image_path = "~/gce_local_image_dir/"
364        # Default to find */avd-system.tar.gz if exist then return the path.
365        self.assertEqual(self.AvdSpec._GetGceLocalImagePath(fake_image_path),
366                         "~/gce_local_image_dir/avd-system.tar.gz")
367
368        # Otherwise choose raw file */android_system_disk_syslinux.img if
369        # exist then return the path.
370        self.Patch(os.path, "exists", side_effect=[False, True])
371        self.assertEqual(self.AvdSpec._GetGceLocalImagePath(fake_image_path),
372                         "~/gce_local_image_dir/android_system_disk_syslinux.img")
373
374        # Both _GCE_LOCAL_IMAGE_CANDIDATE could not be found then raise error.
375        self.Patch(os.path, "exists", side_effect=[False, False])
376        self.assertRaises(errors.ImgDoesNotExist,
377                          self.AvdSpec._GetGceLocalImagePath, fake_image_path)
378
379    def testProcessMiscArgs(self):
380        """Test process misc args."""
381        self.args.remote_host = None
382        self.args.local_instance = None
383        self.AvdSpec._ProcessMiscArgs(self.args)
384        self.assertEqual(self.AvdSpec._instance_type, constants.INSTANCE_TYPE_REMOTE)
385
386        self.args.remote_host = None
387        self.args.local_instance = True
388        self.AvdSpec._ProcessMiscArgs(self.args)
389        self.assertEqual(self.AvdSpec._instance_type, constants.INSTANCE_TYPE_LOCAL)
390
391        self.args.remote_host = "1.1.1.1"
392        self.args.local_instance = None
393        self.AvdSpec._ProcessMiscArgs(self.args)
394        self.assertEqual(self.AvdSpec._instance_type, constants.INSTANCE_TYPE_HOST)
395
396        self.args.remote_host = "1.1.1.1"
397        self.args.local_instance = True
398        self.AvdSpec._ProcessMiscArgs(self.args)
399        self.assertEqual(self.AvdSpec._instance_type, constants.INSTANCE_TYPE_HOST)
400
401        # Test avd_spec.autoconnect
402        self.args.autoconnect = False
403        self.AvdSpec._ProcessMiscArgs(self.args)
404        self.assertEqual(self.AvdSpec.autoconnect, False)
405        self.assertEqual(self.AvdSpec.connect_adb, False)
406        self.assertEqual(self.AvdSpec.connect_vnc, False)
407        self.assertEqual(self.AvdSpec.connect_webrtc, False)
408
409        self.args.autoconnect = constants.INS_KEY_VNC
410        self.AvdSpec._ProcessMiscArgs(self.args)
411        self.assertEqual(self.AvdSpec.autoconnect, True)
412        self.assertEqual(self.AvdSpec.connect_adb, True)
413        self.assertEqual(self.AvdSpec.connect_vnc, True)
414        self.assertEqual(self.AvdSpec.connect_webrtc, False)
415
416        self.args.autoconnect = constants.INS_KEY_ADB
417        self.AvdSpec._ProcessMiscArgs(self.args)
418        self.assertEqual(self.AvdSpec.autoconnect, True)
419        self.assertEqual(self.AvdSpec.connect_adb, True)
420        self.assertEqual(self.AvdSpec.connect_vnc, False)
421        self.assertEqual(self.AvdSpec.connect_webrtc, False)
422
423        self.args.autoconnect = constants.INS_KEY_WEBRTC
424        self.AvdSpec._ProcessMiscArgs(self.args)
425        self.assertEqual(self.AvdSpec.autoconnect, True)
426        self.assertEqual(self.AvdSpec.connect_adb, True)
427        self.assertEqual(self.AvdSpec.connect_vnc, False)
428        self.assertEqual(self.AvdSpec.connect_webrtc, True)
429
430
431if __name__ == "__main__":
432    unittest.main()
433