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
|
#!/usr/bin/python -S
#
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
# pylint: disable-msg=W0403
"""Update stacked_on_location for all Bazaar branches.
Expects standard input of:
'<id> <branch_type> <unique_name> <stacked_on_id> <stacked_on_unique_name>\n'.
Such input can be provided using "get-stacked-on-branches.py".
This script makes the stacked_on_location variables in all Bazaar branches
match the stacked_on column in the Launchpad database. This is useful for
updating stacked branches when their stacked-on branch has been moved or
renamed.
"""
__metaclass__ = type
import _pythonpath
from collections import namedtuple
import sys
from bzrlib import errors
from bzrlib.bzrdir import BzrDir
from bzrlib.config import TransportConfig
from lp.code.interfaces.codehosting import branch_id_alias
from lp.codehosting.bzrutils import get_branch_stacked_on_url
from lp.codehosting.vfs import (
get_ro_server,
get_rw_server,
)
from lp.services.scripts.base import LaunchpadScript
FakeBranch = namedtuple('FakeBranch', 'id')
def set_branch_stacked_on_url(bzrdir, stacked_on_url):
"""Set the stacked_on_location for the branch at 'bzrdir'.
We cannot use Branch.set_stacked_on, since that requires us to first open
the branch. Opening the branch requires a working stacked_on_url:
something we don't yet have.
"""
branch_transport = bzrdir.get_branch_transport(None)
branch_config = TransportConfig(branch_transport, 'branch.conf')
stacked_on_url = branch_config.set_option(
stacked_on_url, 'stacked_on_location')
class UpdateStackedBranches(LaunchpadScript):
"""Update stacked branches so their stacked_on_location matches the db."""
def __init__(self):
super(UpdateStackedBranches, self).__init__('update-stacked-on')
def add_my_options(self):
self.parser.add_option(
'-n', '--dry-run', default=False, action="store_true",
dest="dry_run",
help=("Don't change anything on disk, just go through the "
"motions."))
self.parser.add_option(
'-i', '--id', default=False, action="store_true",
dest="stack_on_id",
help=("Stack on the +branch-id alias."))
def main(self):
if self.options.dry_run:
server = get_ro_server()
else:
server = get_rw_server()
server.start_server()
if self.options.dry_run:
self.logger.debug('Running read-only')
self.logger.debug('Beginning processing')
try:
self.updateBranches(self.parseFromStream(sys.stdin))
finally:
server.stop_server()
self.logger.info('Done')
def updateStackedOn(self, branch_id, bzr_branch_url, stacked_on_location):
"""Stack the Bazaar branch at 'bzr_branch_url' on the given URL.
:param branch_id: The database ID of the branch. This is only used for
logging.
:param bzr_branch_url: The lp-internal:/// URL of the Bazaar branch.
:param stacked_on_location: The location to store in the branch's
stacked_on_location configuration variable.
"""
try:
bzrdir = BzrDir.open(bzr_branch_url)
except errors.NotBranchError:
self.logger.warn(
"No bzrdir for %r at %r" % (branch_id, bzr_branch_url))
return
try:
current_stacked_on_location = get_branch_stacked_on_url(bzrdir)
except errors.NotBranchError:
self.logger.warn(
"No branch for %r at %r" % (branch_id, bzr_branch_url))
except errors.NotStacked:
self.logger.warn(
"Branch for %r at %r is not stacked at all. Giving up."
% (branch_id, bzr_branch_url))
except errors.UnstackableBranchFormat:
self.logger.error(
"Branch for %r at %r is unstackable. Giving up."
% (branch_id, bzr_branch_url))
else:
if current_stacked_on_location != stacked_on_location:
self.logger.info(
'Branch for %r at %r stacked on %r, should be on %r.'
% (branch_id, bzr_branch_url, current_stacked_on_location,
stacked_on_location))
if not self.options.dry_run:
set_branch_stacked_on_url(bzrdir, stacked_on_location)
def parseFromStream(self, stream):
"""Parse branch input from the given stream.
Expects the stream to be populated only by blank lines or by lines
with whitespace-separated fields. Such lines are yielded as tuples.
Blank lines are ignored.
"""
for line in stream.readlines():
if not line.strip():
continue
yield line.split()
def updateBranches(self, branches):
"""Update the stacked_on_location for all branches in 'branches'.
:param branches: An iterator yielding (branch_id, branch_type,
unique_name, stacked_on_unique_name).
"""
for branch_info in branches:
(branch_id, branch_type, unique_name,
stacked_on_id, stacked_on_name) = branch_info
if self.options.stack_on_id:
branch = FakeBranch(stacked_on_id)
stacked_on_location = branch_id_alias(branch)
else:
stacked_on_location = '/' + stacked_on_name
self.updateStackedOn(
branch_id, 'lp-internal:///' + unique_name,
stacked_on_location)
if __name__ == '__main__':
UpdateStackedBranches().lock_and_run()
|