| #!/usr/bin/env python2.7 | |
| # Copyright 2019 The Fuchsia Authors. All rights reserved. | |
| # Use of this source code is governed by a BSD-style license that can be | |
| # found in the LICENSE file. | |
| import os | |
| import unittest | |
| import subprocess | |
| import test_env | |
| from lib.device import Device | |
| from test_case import TestCaseWithFactory | |
| class DeviceTest(TestCaseWithFactory): | |
| # These tests don't use a factory-created Device. | |
| # This to avoid having the factory add SSH options. | |
| def test_ssh_config(self): | |
| device = Device(self.factory, addr='::1') | |
| self.assertFalse(device.ssh_config) | |
| with self.assertRaises(ValueError): | |
| device.ssh_config = 'no_such_config' | |
| ssh_config = 'ssh_config' | |
| self.host.touch(ssh_config) | |
| device.ssh_config = ssh_config | |
| self.assertEqual(['-F', ssh_config], device.ssh_opts()) | |
| def test_ssh_identity(self): | |
| device = Device(self.factory, addr='::1') | |
| self.assertFalse(device.ssh_identity) | |
| with self.assertRaises(ValueError): | |
| device.ssh_identity = 'no_such_identity' | |
| ssh_identity = 'ssh_identity' | |
| self.host.touch(ssh_identity) | |
| device.ssh_identity = ssh_identity | |
| self.assertEqual(['-i', ssh_identity], device.ssh_opts()) | |
| def test_ssh_option(self): | |
| device = Device(self.factory, addr='::1') | |
| self.assertEqual(device.ssh_options, []) | |
| device.ssh_options = ['StrictHostKeyChecking no'] | |
| device.ssh_options.append('UserKnownHostsFile=/dev/null') | |
| self.assertEqual( | |
| [ | |
| '-o', | |
| 'StrictHostKeyChecking no', | |
| '-o', | |
| 'UserKnownHostsFile=/dev/null', | |
| ], device.ssh_opts()) | |
| def test_ssh_verbosity(self): | |
| device = Device(self.factory, addr='::1') | |
| self.assertEqual(device.ssh_verbosity, 0) | |
| with self.assertRaises(ValueError): | |
| device.ssh_verbosity = 4 | |
| device.ssh_verbosity = 3 | |
| self.assertEqual(['-vvv'], device.ssh_opts()) | |
| device.ssh_verbosity = 2 | |
| self.assertEqual(['-vv'], device.ssh_opts()) | |
| device.ssh_verbosity = 1 | |
| self.assertEqual(['-v'], device.ssh_opts()) | |
| device.ssh_verbosity = 0 | |
| self.assertEqual([], device.ssh_opts()) | |
| with self.assertRaises(ValueError): | |
| device.ssh_verbosity = -1 | |
| def test_reachable(self): | |
| cmd = ['true'] | |
| self.assertTrue(self.device.reachable) | |
| self.assertSsh(*cmd) | |
| def test_unreachable(self): | |
| cmd = ['true'] | |
| process = self.get_process(cmd, ssh=True) | |
| process.succeeds = False | |
| self.assertFalse(self.device.reachable) | |
| self.assertSsh(*cmd) | |
| def test_configure(self): | |
| device = Device(self.factory, addr='::1') | |
| device.configure() | |
| self.assertEqual( | |
| device.ssh_config, | |
| self.buildenv.abspath( | |
| self.buildenv.build_dir, 'ssh-keys', 'ssh_config')) | |
| self.assertFalse(device.ssh_identity) | |
| self.assertFalse(device.ssh_options) | |
| self.assertFalse(device.ssh_verbosity) | |
| # These tests use a Device created by the TestCase | |
| def test_ssh(self): | |
| cmd = ['some-command', '--with', 'some-argument'] | |
| self.device.ssh(cmd).check_call() | |
| self.assertSsh(*cmd) | |
| def test_has_cs_info(self): | |
| url1 = 'fuchsia-pkg://fuchsia.com/http#meta/http.cmx' | |
| url2 = 'fuchsia-pkg://fuchsia.com/fake-package1#meta/fake-target1.cmx' | |
| url3 = 'fuchsia-pkg://fuchsia.com/fake-package1#meta/fake-target2.cmx' | |
| url4 = 'fuchsia-pkg://fuchsia.com/an-extremely-verbose-target-package#meta/an-extremely-verbose-target-executable.cmx' | |
| self.set_running(url1) | |
| self.set_running(url2, duration=10) | |
| self.set_running(url3, duration=10) | |
| self.set_running(url4) | |
| # Can check various URLs | |
| self.assertTrue(self.device.has_cs_info(url1)) | |
| self.assertTrue(self.device.has_cs_info(url2)) | |
| self.assertTrue(self.device.has_cs_info(url3)) | |
| self.assertTrue(self.device.has_cs_info(url4)) | |
| # URLs are cached until refresh. | |
| self.host.sleep(10) | |
| self.assertTrue(self.device.has_cs_info(url1)) | |
| self.assertTrue(self.device.has_cs_info(url2)) | |
| self.assertTrue(self.device.has_cs_info(url3)) | |
| self.assertTrue(self.device.has_cs_info(url4)) | |
| self.assertTrue(self.device.has_cs_info(url1, refresh=True)) | |
| self.assertFalse(self.device.has_cs_info(url2)) | |
| self.assertFalse(self.device.has_cs_info(url3)) | |
| self.assertTrue(self.device.has_cs_info(url4)) | |
| def test_isfile(self): | |
| some_file = 'path-to-some-file' | |
| cmd = ['test', '-f', some_file] | |
| process = self.get_process(cmd, ssh=True) | |
| process.succeeds = False | |
| self.assertFalse(self.device.isfile(some_file)) | |
| self.assertSsh(*cmd) | |
| process.succeeds = True | |
| self.assertTrue(self.device.isfile(some_file)) | |
| self.assertSsh(*cmd) | |
| def test_isdir(self): | |
| some_dir = 'path-to-some-directory' | |
| cmd = ['test', '-d', some_dir] | |
| process = self.get_process(cmd, ssh=True) | |
| process.succeeds = False | |
| self.assertFalse(self.device.isdir(some_dir)) | |
| self.assertSsh(*cmd) | |
| process.succeeds = True | |
| self.assertTrue(self.device.isdir(some_dir)) | |
| self.assertSsh(*cmd) | |
| def test_ls(self): | |
| corpus_dir = 'path-to-some-corpus' | |
| cmd = ['ls', '-l', corpus_dir] | |
| self.touch_on_device( | |
| '{}/feac37187e77ff60222325cf2829e2273e04f2ea'.format(corpus_dir), | |
| size=1796) | |
| self.touch_on_device( | |
| '{}/ff415bddb30e9904bccbbd21fb5d4aa9bae9e5a5'.format(corpus_dir), | |
| size=124) | |
| files = self.device.ls(corpus_dir) | |
| self.assertSsh(*cmd) | |
| self.assertEqual( | |
| files['feac37187e77ff60222325cf2829e2273e04f2ea'], 1796) | |
| def test_mkdir(self): | |
| corpus_dir = 'path-to-some-corpus' | |
| cmd = ['mkdir', '-p', corpus_dir] | |
| self.device.mkdir(corpus_dir) | |
| self.assertSsh(*cmd) | |
| def test_remove(self): | |
| some_file = 'path-to-some-file' | |
| cmd = ['rm', '-f', some_file] | |
| self.device.remove(some_file) | |
| self.assertSsh(*cmd) | |
| some_dir = 'path-to-some-directory' | |
| cmd = ['rm', '-rf', some_dir] | |
| self.device.remove(some_dir, recursive=True) | |
| self.assertSsh(*cmd) | |
| def test_dump_log(self): | |
| cmd = [ | |
| 'log_listener', '--dump_logs', 'yes', '--pretty', 'no', '--some', | |
| 'other-arg' | |
| ] | |
| self.device.dump_log('--some', 'other-arg') | |
| self.assertSsh(*cmd) | |
| def test_guess_pid(self): | |
| cmd = [ | |
| 'log_listener', '--dump_logs', 'yes', '--pretty', 'no', '--only', | |
| 'reset,Fuzzer,Sanitizer' | |
| ] | |
| self.assertEqual(self.device.guess_pid(), -1) | |
| self.assertSsh(*cmd) | |
| # Log lines are like '[timestamp][pid][tid][name] data' | |
| self.set_outputs( | |
| cmd, [ | |
| '[ignored][31415][ignored][ignored] ignored', | |
| '[ignored][92653][ignored][ignored] ignored', | |
| '[malformed][58979]', | |
| ], | |
| ssh=True) | |
| # Should find last well-formed line | |
| self.assertEqual(self.device.guess_pid(), 92653) | |
| def test_fetch(self): | |
| local_path = 'test_fetch' | |
| remote_path = 'remote-path' | |
| # Fails due to missing local pathname. | |
| self.assertError( | |
| lambda: self.device.fetch(local_path, remote_path), | |
| 'No such directory: test_fetch') | |
| self.host.mkdir(local_path) | |
| # Fails due to empty source file list. | |
| with self.assertRaises(ValueError): | |
| self.device.fetch(local_path) | |
| self.device.fetch(local_path, remote_path) | |
| self.assertScpFrom(remote_path, local_path) | |
| corpus_dir = remote_path + '/*' | |
| self.device.fetch(local_path, corpus_dir) | |
| self.assertScpFrom(corpus_dir, local_path) | |
| def test_store(self): | |
| local_path = 'test_store' | |
| remote_path = 'remote-path' | |
| # Globs must resolve | |
| self.assertError( | |
| lambda: self.device.store( | |
| remote_path, os.path.join(local_path, '*')), | |
| 'No matching files: "test_store/*".') | |
| # Local path must exist | |
| foo = os.path.join(local_path, 'foo') | |
| self.assertError( | |
| lambda: self.device.store(remote_path, foo), | |
| 'No matching files: "test_store/foo".') | |
| # Valid | |
| self.host.touch(foo) | |
| self.device.store(remote_path, foo) | |
| self.assertScpTo(foo, remote_path) | |
| # Valid globs | |
| bar = os.path.join(local_path, 'bar') | |
| baz = os.path.join(local_path, 'baz') | |
| self.host.touch(bar) | |
| self.host.touch(baz) | |
| self.device.store(remote_path, os.path.join(local_path, '*')) | |
| self.assertScpTo(bar, baz, foo, remote_path) | |
| if __name__ == '__main__': | |
| unittest.main() |