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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Server used in codehosting acceptance tests."""
__metaclass__ = type
__all__ = [
'CodeHostingTac',
'SSHCodeHostingServer',
]
import os
import shutil
import tempfile
from bzrlib.transport import (
get_transport,
Server,
)
from twisted.python.util import sibpath
from zope.component import getUtility
from canonical.config import config
from canonical.database.sqlbase import commit
from lp.services.daemons.tachandler import TacTestSetup
from lp.registry.interfaces.person import (
IPersonSet,
TeamSubscriptionPolicy,
)
from lp.registry.interfaces.ssh import ISSHKeySet
def set_up_test_user(test_user, test_team):
"""Configure a user called 'test_user' with SSH keys.
Also make sure that 'test_user' belongs to 'test_team'.
"""
person_set = getUtility(IPersonSet)
testUser = person_set.getByName('no-priv')
testUser.name = test_user
testTeam = person_set.newTeam(
testUser, test_team, test_team,
subscriptionpolicy=TeamSubscriptionPolicy.OPEN)
testUser.join(testTeam)
ssh_key_set = getUtility(ISSHKeySet)
ssh_key_set.new(
testUser,
'ssh-dss AAAAB3NzaC1kc3MAAABBAL5VoWG5sy3CnLYeOw47L8m9A15hA/PzdX2u'
'0B7c2Z1ktFPcEaEuKbLqKVSkXpYm7YwKj9y88A9Qm61CdvI0c50AAAAVAKGY0YON'
'9dEFH3DzeVYHVEBGFGfVAAAAQCoe0RhBcefm4YiyQVwMAxwTlgySTk7FSk6GZ95E'
'Z5Q8/OTdViTaalvGXaRIsBdaQamHEBB+Vek/VpnF1UGGm8YAAABAaCXDl0r1k93J'
'hnMdF0ap4UJQ2/NnqCyoE8Xd5KdUWWwqwGdMzqB1NOeKN6ladIAXRggLc2E00Usn'
'UXh3GE3Rgw== testuser')
commit()
class CodeHostingTac(TacTestSetup):
def __init__(self, mirrored_area):
super(CodeHostingTac, self).__init__()
# The mirrored area.
self._mirror_root = mirrored_area
# Where the pidfile, logfile etc will go.
self._server_root = tempfile.mkdtemp()
def clear(self):
"""Clear the branch areas."""
if os.path.isdir(self._mirror_root):
shutil.rmtree(self._mirror_root)
os.makedirs(self._mirror_root, 0700)
def setUpRoot(self):
self.clear()
def tearDownRoot(self):
shutil.rmtree(self._server_root)
@property
def root(self):
return self._server_root
@property
def tacfile(self):
return os.path.abspath(
os.path.join(config.root, 'daemons', 'sftp.tac'))
@property
def logfile(self):
return os.path.join(self.root, 'codehosting.log')
@property
def pidfile(self):
return os.path.join(self.root, 'codehosting.pid')
class SSHCodeHostingServer(Server):
def __init__(self, schema, tac_server):
Server.__init__(self)
self._schema = schema
# XXX: JonathanLange 2008-10-08: This is used by createBazaarBranch in
# test_acceptance.
self._mirror_root = tac_server._mirror_root
self._tac_server = tac_server
def setUpFakeHome(self):
user_home = os.path.abspath(tempfile.mkdtemp())
os.makedirs(os.path.join(user_home, '.ssh'))
shutil.copyfile(
sibpath(__file__, 'id_dsa'),
os.path.join(user_home, '.ssh', 'id_dsa'))
shutil.copyfile(
sibpath(__file__, 'id_dsa.pub'),
os.path.join(user_home, '.ssh', 'id_dsa.pub'))
os.chmod(os.path.join(user_home, '.ssh', 'id_dsa'), 0600)
real_home, os.environ['HOME'] = os.environ['HOME'], user_home
return real_home, user_home
def getTransport(self, path=None):
if path is None:
path = ''
transport = get_transport(self.get_url()).clone(path)
return transport
def start_server(self):
self._real_home, self._fake_home = self.setUpFakeHome()
def stop_server(self):
os.environ['HOME'] = self._real_home
shutil.rmtree(self._fake_home)
def get_url(self, user=None):
if user is None:
user = 'testuser'
return '%s://%s@bazaar.launchpad.dev:22222/' % (self._schema, user)
|