| #!/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 tempfile |
| |
| import test_env |
| from lib.args import ArgParser |
| from lib.device import Device |
| from lib.host import Host |
| |
| from device_mock import MockDevice |
| from host_mock import MockHost |
| |
| |
| class TestDevice(unittest.TestCase): |
| """ Tests lib.Device. See MockDevice for additional details.""" |
| |
| def test_from_args(self): |
| host = MockHost() |
| parser = ArgParser('description') |
| parser.require_name(False) |
| # netaddr should get called with 'just-four-random-words', and fail |
| with self.assertRaises(RuntimeError): |
| args = parser.parse_args(['--device', 'just-four-random-words']) |
| device = Device.from_args(host, args) |
| |
| def test_set_ssh_config(self): |
| mock = MockDevice() |
| with self.assertRaises(Host.ConfigError): |
| mock.set_ssh_config('no_such_config') |
| if not os.getenv('FUCHSIA_DIR'): |
| return |
| build_dir = mock.host.find_build_dir() |
| ssh_config = Host.join(build_dir, 'ssh-keys', 'ssh_config') |
| if not os.path.exists(ssh_config): |
| return |
| mock.set_ssh_config(ssh_config) |
| cmd = ' '.join(mock.get_ssh_cmd(['ssh'])) |
| self.assertIn('ssh', cmd) |
| self.assertIn(' -F ' + ssh_config, cmd) |
| |
| def test_set_ssh_identity(self): |
| mock = MockDevice() |
| with self.assertRaises(Host.ConfigError): |
| mock.set_ssh_identity('no_such_identity') |
| if not os.getenv('FUCHSIA_DIR'): |
| return |
| identity_file = Host.join('.ssh', 'pkey') |
| if not os.path.exists(identity_file): |
| return |
| mock.set_ssh_identity(identity_file) |
| cmd = ' '.join(mock.get_ssh_cmd(['scp'])) |
| self.assertIn('scp', cmd) |
| self.assertIn(' -i ' + identity_file, cmd) |
| |
| def test_set_ssh_option(self): |
| mock = MockDevice() |
| mock.set_ssh_option('StrictHostKeyChecking no') |
| mock.set_ssh_option('UserKnownHostsFile=/dev/null') |
| cmd = ' '.join(mock.get_ssh_cmd(['ssh'])) |
| self.assertIn(' -o StrictHostKeyChecking no', cmd) |
| self.assertIn(' -o UserKnownHostsFile=/dev/null', cmd) |
| |
| def test_init(self): |
| mock = MockDevice(port=51823) |
| cmd = ' '.join(mock.get_ssh_cmd(['ssh', '::1', 'some-command'])) |
| self.assertIn(' -p 51823 ', cmd) |
| |
| def test_set_ssh_verbosity(self): |
| mock = MockDevice() |
| mock.set_ssh_verbosity(3) |
| cmd = ' '.join(mock.get_ssh_cmd(['ssh', '::1', 'some-command'])) |
| self.assertIn(' -vvv ', cmd) |
| mock.set_ssh_verbosity(1) |
| cmd = ' '.join(mock.get_ssh_cmd(['ssh', '::1', 'some-command'])) |
| self.assertIn(' -v ', cmd) |
| self.assertNotIn(' -vvv ', cmd) |
| mock.set_ssh_verbosity(0) |
| cmd = ' '.join(mock.get_ssh_cmd(['ssh', '::1', 'some-command'])) |
| self.assertNotIn(' -v ', cmd) |
| |
| def test_ssh(self): |
| mock = MockDevice() |
| mock.ssh(['some-command', '--with', 'some-argument']).check_call() |
| self.assertIn( |
| ' '.join( |
| mock.get_ssh_cmd( |
| ['ssh', '::1', 'some-command', '--with some-argument'])), |
| mock.host.history) |
| |
| def test_getpids(self): |
| mock = MockDevice() |
| pids = mock.getpids() |
| self.assertTrue('mock-target1' in pids) |
| self.assertEqual(pids['mock-target1'], 7412221) |
| self.assertEqual(pids['an-extremely-verbose-target-name'], 7412223) |
| |
| def test_ls(self): |
| mock = MockDevice() |
| files = mock.ls('path-to-some-corpus') |
| self.assertIn( |
| ' '.join( |
| mock.get_ssh_cmd( |
| ['ssh', '::1', 'ls', '-l', 'path-to-some-corpus'])), |
| mock.host.history) |
| self.assertTrue('feac37187e77ff60222325cf2829e2273e04f2ea' in files) |
| self.assertEqual( |
| files['feac37187e77ff60222325cf2829e2273e04f2ea'], 1796) |
| |
| def test_rm(self): |
| mock = MockDevice() |
| mock.rm('path-to-some-file') |
| mock.rm('path-to-some-directory', recursive=True) |
| self.assertIn( |
| ' '.join( |
| mock.get_ssh_cmd( |
| ['ssh', '::1', 'rm', '-f', 'path-to-some-file'])), |
| mock.host.history) |
| self.assertIn( |
| ' '.join( |
| mock.get_ssh_cmd( |
| ['ssh', '::1', 'rm', '-rf', 'path-to-some-directory'])), |
| mock.host.history) |
| |
| def test_fetch(self): |
| mock = MockDevice() |
| with self.assertRaises(ValueError): |
| mock.fetch('foo', 'not-likely-to-be-a-directory') |
| mock.fetch('remote-path', '/tmp') |
| self.assertIn( |
| ' '.join(mock.get_ssh_cmd(['scp', '[::1]:remote-path', '/tmp'])), |
| mock.host.history) |
| mock.fetch('corpus/*', '/tmp') |
| self.assertIn( |
| ' '.join(mock.get_ssh_cmd(['scp', '[::1]:corpus/*', '/tmp'])), |
| mock.host.history) |
| mock.delay = 2 |
| with self.assertRaises(subprocess.CalledProcessError): |
| mock.fetch('delayed', '/tmp') |
| mock.delay = 2 |
| with self.assertRaises(subprocess.CalledProcessError): |
| mock.fetch('delayed', '/tmp', retries=1) |
| mock.delay = 2 |
| mock.fetch('delayed', '/tmp', retries=2) |
| self.assertIn( |
| ' '.join(mock.get_ssh_cmd(['scp', '[::1]:delayed', '/tmp'])), |
| mock.host.history) |
| |
| def test_store(self): |
| mock = MockDevice() |
| mock.store(tempfile.gettempdir(), 'remote-path') |
| self.assertIn( |
| ' '.join( |
| mock.get_ssh_cmd( |
| ['scp', tempfile.gettempdir(), '[::1]:remote-path'])), |
| mock.host.history) |
| # Ensure globbing works |
| mock.host.history = [] |
| with tempfile.NamedTemporaryFile() as f: |
| mock.store('{}/*'.format(tempfile.gettempdir()), 'remote-path') |
| for cmd in mock.host.history: |
| if cmd.startswith('scp'): |
| self.assertIn(f.name, cmd) |
| # No copy if glob comes up empty |
| mock.store(os.path.join(tempfile.gettempdir(), '*'), 'remote-path') |
| self.assertNotIn( |
| ' '.join( |
| mock.get_ssh_cmd( |
| ['scp', tempfile.gettempdir(), '[::1]:remote-path'])), |
| mock.host.history) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
