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
|
# Copyright (C) 2005-2007 Jelmer Vernooij <jelmer@samba.org>
# 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/>.
from bzrlib import urlutils
from bzrlib.errors import NoSuchRevision, NoSuchTag
from bzrlib.tag import BasicTags
from bzrlib.trace import mutter
from bzrlib.plugins.svn import commit, subvertpy, errors as svn_errors, mapping, properties
class SubversionTags(BasicTags):
"""Subversion tags object."""
def __init__(self, branch):
self.branch = branch
self.repository = branch.repository
self._parent_exists = False
def _ensure_tag_parent_exists(self, parent):
if self._parent_exists:
return
assert isinstance(parent, str)
bp_parts = parent.split("/")
existing_bp_parts = commit._check_dirs_exist(
self.repository.transport,
bp_parts, self.repository.get_latest_revnum())
if existing_bp_parts == bp_parts:
self._parent_exists = True
return
commit.create_branch_prefix(self.repository,
self._revprops("Add tags base directory."),
bp_parts, existing_bp_parts)
self._parent_exists = True
def set_tag(self, tag_name, tag_target):
path = self.branch.layout.get_tag_path(tag_name, self.branch.project)
assert isinstance(path, str)
parent = urlutils.dirname(path)
try:
(from_bp, from_revnum, mapping) = self.repository.lookup_revision_id(tag_target, project=self.branch.project)
except NoSuchRevision:
mutter("not setting tag %s; unknown revision %s", tag_name, tag_target)
return
if from_bp == path:
return
self._ensure_tag_parent_exists(parent)
conn = self.repository.transport.get_connection(parent)
deletefirst = (conn.check_path(urlutils.basename(path), self.repository.get_latest_revnum()) != subvertpy.NODE_NONE)
try:
ci = svn_errors.convert_svn_error(conn.get_commit_editor)(
self._revprops("Add tag %s" % tag_name.encode("utf-8"),
{tag_name.encode("utf-8"): tag_target}))
try:
root = ci.open_root()
if deletefirst:
root.delete_entry(urlutils.basename(path))
tag_dir = root.add_directory(urlutils.basename(path), urlutils.join(self.repository.base, from_bp), from_revnum)
tag_dir.close()
root.close()
except:
ci.abort()
raise
ci.close()
finally:
self.repository.transport.add_connection(conn)
def _revprops(self, message, tags_dict=None):
"""Create a revprops dictionary.
Optionally sets bzr:skip to slightly optimize fetching of this revision later.
"""
revprops = {properties.PROP_REVISION_LOG: message, }
if self.repository.transport.has_capability("commit-revprops"):
revprops[mapping.SVN_REVPROP_BZR_SKIP] = ""
if tags_dict is not None:
revprops[mapping.SVN_REVPROP_BZR_TAGS] = mapping.generate_tags_property(tags_dict)
return revprops
def lookup_tag(self, tag_name):
try:
return self.get_tag_dict()[tag_name]
except KeyError:
raise NoSuchTag(tag_name)
def get_tag_dict(self):
return self.repository.find_tags(project=self.branch.project,
layout=self.branch.layout,
mapping=self.branch.mapping,
revnum=self.branch._revnum)
def get_reverse_tag_dict(self):
"""Returns a dict with revisions as keys
and a list of tags for that revision as value"""
d = self.get_tag_dict()
rev = {}
for key in d:
try:
rev[d[key]].append(key)
except KeyError:
rev[d[key]] = [key]
return rev
def delete_tag(self, tag_name):
path = self.branch.layout.get_tag_path(tag_name, self.branch.project)
parent = urlutils.dirname(path)
conn = self.repository.transport.get_connection(parent)
try:
if conn.check_path(urlutils.basename(path), self.repository.get_latest_revnum()) != subvertpy.NODE_DIR:
raise NoSuchTag(tag_name)
ci = svn_errors.convert_svn_error(conn.get_commit_editor)(self._revprops("Remove tag %s" % tag_name.encode("utf-8"),
{tag_name: ""}))
try:
root = ci.open_root()
root.delete_entry(urlutils.basename(path))
root.close()
except:
ci.abort()
raise
ci.close()
finally:
assert not conn.busy
self.repository.transport.add_connection(conn)
def _set_tag_dict(self, dest_dict):
cur_dict = self.get_tag_dict()
for k, v in dest_dict.iteritems():
if cur_dict.get(k) != v:
self.set_tag(k, v)
for k in cur_dict:
if k not in dest_dict:
self.delete_tag(k)
|
