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
|
from mocker import MockerTestCase
from cloudinit import distros
unknown_arch_info = {
'arches': ['default'],
'failsafe': {'primary': 'http://fs-primary-default',
'security': 'http://fs-security-default'}
}
package_mirrors = [
{'arches': ['i386', 'amd64'],
'failsafe': {'primary': 'http://fs-primary-intel',
'security': 'http://fs-security-intel'},
'search': {
'primary': ['http://%(ec2_region)s.ec2/',
'http://%(availability_zone)s.clouds/'],
'security': ['http://security-mirror1-intel',
'http://security-mirror2-intel']}},
{'arches': ['armhf', 'armel'],
'failsafe': {'primary': 'http://fs-primary-arm',
'security': 'http://fs-security-arm'}},
unknown_arch_info
]
gpmi = distros._get_package_mirror_info # pylint: disable=W0212
gapmi = distros._get_arch_package_mirror_info # pylint: disable=W0212
class TestGenericDistro(MockerTestCase):
def return_first(self, mlist):
if not mlist:
return None
return mlist[0]
def return_second(self, mlist):
if not mlist:
return None
return mlist[1]
def return_none(self, _mlist):
return None
def return_last(self, mlist):
if not mlist:
return None
return(mlist[-1])
def setUp(self):
super(TestGenericDistro, self).setUp()
# Make a temp directoy for tests to use.
self.tmp = self.makeDir()
def test_arch_package_mirror_info_unknown(self):
"""for an unknown arch, we should get back that with arch 'default'."""
arch_mirrors = gapmi(package_mirrors, arch="unknown")
self.assertEqual(unknown_arch_info, arch_mirrors)
def test_arch_package_mirror_info_known(self):
arch_mirrors = gapmi(package_mirrors, arch="amd64")
self.assertEqual(package_mirrors[0], arch_mirrors)
def test_get_package_mirror_info_az_ec2(self):
arch_mirrors = gapmi(package_mirrors, arch="amd64")
results = gpmi(arch_mirrors, availability_zone="us-east-1a",
mirror_filter=self.return_first)
self.assertEqual(results,
{'primary': 'http://us-east-1.ec2/',
'security': 'http://security-mirror1-intel'})
results = gpmi(arch_mirrors, availability_zone="us-east-1a",
mirror_filter=self.return_second)
self.assertEqual(results,
{'primary': 'http://us-east-1a.clouds/',
'security': 'http://security-mirror2-intel'})
results = gpmi(arch_mirrors, availability_zone="us-east-1a",
mirror_filter=self.return_none)
self.assertEqual(results, package_mirrors[0]['failsafe'])
def test_get_package_mirror_info_az_non_ec2(self):
arch_mirrors = gapmi(package_mirrors, arch="amd64")
results = gpmi(arch_mirrors, availability_zone="nova.cloudvendor",
mirror_filter=self.return_first)
self.assertEqual(results,
{'primary': 'http://nova.cloudvendor.clouds/',
'security': 'http://security-mirror1-intel'})
results = gpmi(arch_mirrors, availability_zone="nova.cloudvendor",
mirror_filter=self.return_last)
self.assertEqual(results,
{'primary': 'http://nova.cloudvendor.clouds/',
'security': 'http://security-mirror2-intel'})
def test_get_package_mirror_info_none(self):
arch_mirrors = gapmi(package_mirrors, arch="amd64")
# because both search entries here replacement based on
# availability-zone, the filter will be called with an empty list and
# failsafe should be taken.
results = gpmi(arch_mirrors, availability_zone=None,
mirror_filter=self.return_first)
self.assertEqual(results,
{'primary': 'http://fs-primary-intel',
'security': 'http://security-mirror1-intel'})
results = gpmi(arch_mirrors, availability_zone=None,
mirror_filter=self.return_last)
self.assertEqual(results,
{'primary': 'http://fs-primary-intel',
'security': 'http://security-mirror2-intel'})
#def _get_package_mirror_info(mirror_info, availability_zone=None,
# mirror_filter=util.search_for_mirror):
# vi: ts=4 expandtab
|
