1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
#!/usr/bin/env python
#
# Copyright 2016 Google Inc.
#
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import default_flavor
import os
import posixpath
import subprocess
import ssh_devices
"""Utils for running tests remotely over SSH."""
class SSHFlavorUtils(default_flavor.DefaultFlavorUtils):
def __init__(self, *args, **kwargs):
super(SSHFlavorUtils, self).__init__(*args, **kwargs)
slave_info = ssh_devices.SLAVE_INFO.get(self._bot_info.slave_name,
ssh_devices.SLAVE_INFO['default'])
self._host = slave_info.ssh_host
self._port = slave_info.ssh_port
self._user = slave_info.ssh_user
@property
def host(self):
return self._host
@property
def port(self):
return self._port
@property
def user(self):
return self._user
def ssh(self, cmd, **kwargs):
"""Run the given SSH command."""
ssh_cmd = ['ssh']
if self.port:
ssh_cmd.extend(['-p', self.port])
dest = self.host
if self.user:
dest = self.user + '@' + dest
ssh_cmd.append(dest)
ssh_cmd.extend(cmd)
return self._bot_info.run(ssh_cmd, **kwargs)
def step(self, *args, **kwargs):
"""Run the given step over SSH."""
self.ssh(*args, **kwargs)
def device_path_join(self, *args):
"""Like os.path.join(), but for paths on a remote machine."""
return posixpath.join(*args)
def device_path_exists(self, path): # pragma: no cover
"""Like os.path.exists(), but for paths on a remote device."""
try:
self.ssh(['test', '-e', path])
return True
except subprocess.CalledProcessError:
return False
def _remove_device_dir(self, path):
"""Remove the directory on the device."""
self.ssh(['rm', '-rf', path])
def _create_device_dir(self, path):
"""Create the directory on the device."""
self.ssh(['mkdir', '-p', path])
def create_clean_device_dir(self, path):
"""Like shutil.rmtree() + os.makedirs(), but on a remote device."""
self._remove_device_dir(path)
self._create_device_dir(path)
def _make_scp_cmd(self, remote_path, recurse=True):
"""Prepare an SCP command.
Returns a partial SCP command and an adjusted remote path.
"""
cmd = ['scp']
if recurse:
cmd.append('-r')
if self.port:
cmd.extend(['-P', self.port])
adj_remote_path = self.host + ':' + remote_path
if self.user:
adj_remote_path = self.user + '@' + adj_remote_path
return cmd, adj_remote_path
def copy_directory_contents_to_device(self, host_dir, device_dir):
"""Like shutil.copytree(), but for copying to a remote device."""
_, remote_path = self._make_scp_cmd(device_dir)
cmd = [os.path.join(self._bot_info.skia_dir, 'tools',
'scp_dir_contents.sh'),
host_dir, remote_path]
self._bot_info.run(cmd)
def copy_directory_contents_to_host(self, device_dir, host_dir):
"""Like shutil.copytree(), but for copying from a remote device."""
_, remote_path = self._make_scp_cmd(device_dir)
cmd = [os.path.join(self._bot_info.skia_dir, 'tools',
'scp_dir_contents.sh'),
remote_path, host_dir]
self._bot_info.run(cmd)
def copy_file_to_device(self, host_path, device_path):
"""Like shutil.copyfile, but for copying to a connected device."""
cmd, remote_path = self._make_scp_cmd(device_path, recurse=False)
cmd.extend([host_path, remote_path])
self._bot_info.run(cmd)
def read_file_on_device(self, path):
return self.ssh(['cat', path]).rstrip()
def remove_file_on_device(self, path):
"""Delete the given file."""
return self.ssh(['rm', '-f', path])
|
