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
|
import sys
import os
import urllib.parse
import urllib.request, urllib.parse, urllib.error
from MgaRepo import Error, config, layout
from MgaRepo.svn import SVN
def mirror_url():
mirror = config.get("global", "mirror")
return mirror
def normalize_path(url):
"""normalize url for relocate_path needs"""
parsed = urllib.parse.urlparse(url)
path = os.path.normpath(parsed[2])
newurl = urllib.parse.urlunparse((parsed[0], parsed[1], path,
parsed[3], parsed[4], parsed[5]))
return newurl
def _joinurl(url, relpath):
parsed = urllib.parse.urlparse(url)
newpath = os.path.join(parsed[2], relpath)
newurl = urllib.parse.urlunparse((parsed[0], parsed[1], newpath,
parsed[3], parsed[4], parsed[5]))
return newurl
def strip_username(url):
parsed = list(urllib.parse.urlparse(url))
_, parsed[1] = urllib.parse.splituser(parsed[1])
newurl = urllib.parse.urlunparse(parsed)
return newurl
def same_base(parent, url):
"""returns true if parent is parent of url"""
parent = normalize_path(parent)
url = normalize_path(url)
url = strip_username(url)
return url.startswith(parent)
def relocate_path(oldparent, newparent, url):
oldparent = normalize_path(oldparent)
newparent = normalize_path(newparent)
url = normalize_path(url)
subpath = url[len(oldparent)+1:]
newurl = _joinurl(newparent, subpath) # subpath usually gets / at begining
return newurl
def enabled(wcurl=None):
mirror = mirror_url()
repository = layout.repository_url()
enabled = False
if mirror and repository:
enabled = True
if wcurl and not same_base(mirror, wcurl):
enabled = False
return enabled
def using_on(url):
"""returnes True if the URL points to the mirror repository"""
mirror = mirror_url()
if mirror:
using = same_base(mirror, url)
else:
using = False
return using
def info(url, write=False, stream=sys.stderr):
if using_on(url):
stream.write("Using the svn mirror.\n")
if write:
stream.write("To be able to commit changes, use "
"'mgarepo switch' first.\n")
def mirror_relocate(oldparent, newparent, url, wcpath):
svn = SVN()
newurl = relocate_path(oldparent, newparent, url)
svn.switch(newurl, url, path=wcpath, relocate=True)
return newurl
def switchto_parent(svn, url, path):
"""Relocates the working copy to default_parent"""
newurl = mirror_relocate(mirror_url(), layout.repository_url(), url, path)
return newurl
def switchto_parent_url(url):
newurl = relocate_path(mirror_url(), layout.repository_url(), url)
return newurl
def switchto_mirror(svn, url, path):
newurl = mirror_relocate(layout.repository_url(), mirror_url(), url, path)
return newurl
def autoswitch(svn, wcpath, wcurl, newbaseurl=None):
"""Switches between mirror, default_parent, or newbaseurl"""
nobase = False
mirror = mirror_url()
repository = layout.repository_url()
current = repository
if repository is None:
raise Error("the option repository from mgarepo.conf is "\
"required")
indefault = same_base(repository, wcurl)
if not newbaseurl:
if not mirror:
raise Error("an URL is needed when the option mirror "\
"from mgarepo.conf is not set")
if indefault:
chosen = mirror
elif same_base(mirror, wcurl):
current = mirror
chosen = repository
else:
nobase = True
else:
if mirror and same_base(mirror, wcurl):
current = mirror
elif indefault:
pass # !!!!
else:
nobase = True
chosen = newbaseurl
if nobase:
raise Error("the URL of this working copy is not based in "\
"repository nor mirror URLs")
assert current != chosen
newurl = mirror_relocate(current, chosen, wcurl, wcpath)
return newurl
|
