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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
# Copyright (C) 2020 Codethink Limited
#
# 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; version 2 of the License.
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
import abc
import shlex
import urllib.parse
import cliapp
class DownstreamHost(abc.ABC):
@staticmethod
def add_app_settings(app_settings):
'''Add any application settings that are specific to this Downstream
Host type.
'''
pass
@staticmethod
def check_app_settings(app_settings):
'''Validate any fields in the application settings that are specific
to this Downstream Host type.
'''
pass
@abc.abstractmethod
def __init__(self, app_settings):
'''Construct a Downstream Host connector from the application
settings.
'''
pass
@abc.abstractmethod
def prepare_repo(self, repo_path, metadata):
'''Prepare a repository on the Host. If the repository does not
exist, this method must create it. It should also set any
given metadata on the repository, whether or not it already
exists.
repo_path is the path that the repository should appear at
within the Host.
metadata is a dictionary with the following (optional) keys
defined:
- head: Name of the default branch (a.k.a. HEAD)
- description: Short string describing the repository
'''
pass
class UpstreamHost(abc.ABC):
@staticmethod
def check_host_type_params(validator, section):
'''Validate any type-specific fields in a CONFGIT host section.
validator is an instance of LorryControllerConfValidator that
may be used to check the types of configuration fields.
section is the dictionary of fields for the section.
Returns None if the configuration is valid; raises an
exception on error.
'''
pass
@staticmethod
def get_host_type_params(section):
'''Convert any type-specific fields in a CONFGIT host section into a
dictionary that will be stored in STATEDB.
section is the dictionary of fields for the section.
Returns a dictionary, which may be empty. This will be stored
in STATEDB as the type_params of the host.
'''
return {}
@abc.abstractmethod
def __init__(self, host_info):
'''Construct an Upstream Host connector from the given host_info.
The host_info comes directly from STATEDB.
'''
pass
@abc.abstractmethod
def list_repos(self):
'''List all visible repositories on the Host.
Returns a list of path strings.
'''
pass
@abc.abstractmethod
def get_repo_url(self, repo_path):
'''Get URL for a repository.
repo_path is the path to the repository within the Host.
Returns a URL string suitable for passing to git clone.
'''
pass
@abc.abstractmethod
def get_repo_metadata(self, repo_path):
'''Get metadata for a repository.
repo_path is the path to the repository within the Host.
Returns a dictionary of metadata suitable for passing to
DownstreamHost.prepare_repo.
'''
pass
class SshCommand:
def __init__(self, urlstring, **options):
try:
url = urllib.parse.urlsplit(urlstring, allow_fragments=False)
url.port
except ValueError:
raise cliapp.AppException('Invalid URL: %s' % urlstring)
if url.scheme != 'ssh':
raise cliapp.AppException('Not an SSH URL: %s' % urlstring)
if url.path not in ['', '/']:
raise cliapp.AppException('Unexpected path part in SSH URL')
if url.query != '':
raise cliapp.AppException('Unexpected query part in SSH URL')
if url.password is not None:
raise cliapp.AppException('Unexpected password in SSH URL')
self._ssh_args = ['ssh', '-oBatchMode=yes']
for key, value in options.items():
self._ssh_args.append('-o%s=%s' % (key, value))
if url.username is not None:
self._ssh_args.append('-oUser=%s' % url.username)
if url.port is not None:
self._ssh_args.append('-p%i' % url.port)
self._ssh_args.append(url.hostname)
def run(self, args):
quoted_args = [shlex.quote(arg) for arg in args]
stdout = cliapp.runcmd(self._ssh_args + quoted_args)
if isinstance(stdout, bytes):
stdout = stdout.decode('utf-8', errors='replace')
return stdout
|
