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
|
# Copyright 2015 Lars Wirzenius
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# =*= License: GPL-3+ =*=
import os
import cliapp
class Target(object):
def __init__(self, name):
self.name = name
self.address = None
self.pbuilder_ci_tgz = None
self.tmpdir = None
def set_address(self, address):
self.address = address
def set_pbuilder_ci_tgz(self, tgz):
self.pbuilder_ci_tgz = tgz
def set_tmpdir(self, tmpdir): # pragma: no cover
self.tmpdir = tmpdir
def sync_to_target(self, srcdir, r_dir, r_subdir): # pragma: no cover
remote_path = os.path.join(r_dir, r_subdir)
self.ssh_runcmd(['mkdir', '-p', remote_path])
cliapp.runcmd(
['rsync', '-asHS', '--delete',
srcdir + '/.',
'%s:%s/.' % (self.address, remote_path)])
return remote_path
def sync_from_target(self, remote_dir, local_dir): # pragma: no cover
cliapp.runcmd(
['rsync', '-asHS', '--delete',
'%s:%s/.' % (self.address, remote_dir),
local_dir + '/.'])
def mkdtemp(self): # pragma: no cover
'''Create a temporary directory on target.'''
prefix = []
if self.tmpdir is not None:
prefix = ['env', 'TMPDIR=%s' % self.tmpdir]
output = self.ssh_runcmd(prefix + ['mktemp', '-d'])
return output.rstrip('\n')
def remove_all(self, pathnames): # pragma: no cover
assert type(pathnames) is list
self.ssh_runcmd(['rm', '-rf'] + pathnames)
def ssh_runcmd(self, *args, **kwargs): # pragma: no cover
return cliapp.ssh_runcmd(self.address, *args, **kwargs)
def create_targets_from_ick(ick, wanted_names):
targets = []
if not wanted_names:
wanted_names = ick.get('targets', {}).keys()
for name, spec in ick.get('targets', {}).items():
if name in wanted_names:
t = Target(name)
t.set_address(spec['address'])
t.set_pbuilder_ci_tgz(
spec.get(
'pbuilder-ci-tgz',
'/var/cache/pbuilder/base.tgz'))
if 'tmpdir' in spec: # pragma: no cover
t.set_tmpdir(spec['tmpdir'])
for x in ['debian_arch', 'debian_codename', 'debian_release']:
if x in spec:
setattr(t, x, spec[x]) # pragma: no cover
targets.append(t)
return targets
|
