← Back to branch summary

~andrewjbeach/juju-ci-tools/make-local-patcher

« back to all changes in this revision

Viewing changes to tests/test_update_lxc_cache.py

  • Committer: Aaron Bentley
  • Date: 2014-02-28 16:40:22 UTC
  • mto: This revision was merged to the branch mainline in revision 257.
  • Revision ID: aaron.bentley@canonical.com-20140228164022-kfip2tphn9m9invi
Add juju-backup script.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_update_lxc_cache.py
1
 
from mock import patch
2
 
import os
3
 
from StringIO import StringIO
4
 
from unittest import TestCase
5
 
 
6
 
from update_lxc_cache import (
7
 
    INDEX,
8
 
    INDEX_PATH,
9
 
    INSTALL_SCRIPT,
10
 
    LxcCache,
11
 
    main,
12
 
    parse_args,
13
 
    PUT_SCRIPT,
14
 
    System
15
 
    )
16
 
from utility import temp_dir
17
 
 
18
 
 
19
 
INDEX_DATA = """\
20
 
ubuntu;trusty;arm64;default;201501;/images/ubuntu/trusty/arm64/default/201501/
21
 
ubuntu;trusty;armhf;default;201502;/images/ubuntu/trusty/armhf/default/201502/
22
 
ubuntu;trusty;i386;default;201503;/images/ubuntu/trusty/i386/default/201503/
23
 
ubuntu;trusty;ppc64el;default;20154;/images/ubuntu/trusty/ppc64el/default/20154/
24
 
"""
25
 
 
26
 
 
27
 
def make_systems(data=INDEX_DATA):
28
 
    systems = {}
29
 
    for line in data.splitlines():
30
 
        system = System(*line.split(';'))
31
 
        key = (system.dist, system.release, system.arch, system.variant)
32
 
        systems[key] = system
33
 
    return systems
34
 
 
35
 
 
36
 
def make_local_cache(workspace):
37
 
    meta_dir = os.path.join(workspace, INDEX_PATH)
38
 
    os.makedirs(meta_dir)
39
 
    index_path = os.path.join(meta_dir, INDEX)
40
 
    with open(index_path, 'w') as f:
41
 
        f.write(INDEX_DATA)
42
 
    return index_path
43
 
 
44
 
 
45
 
class FakeResponse(StringIO):
46
 
 
47
 
    def __init__(self, code, data, *args, **kwargs):
48
 
        StringIO.__init__(self, *args, **kwargs)
49
 
        self.code = code
50
 
        self.write(data)
51
 
        self.seek(0)
52
 
 
53
 
    def getcode(self):
54
 
        return self.code
55
 
 
56
 
 
57
 
class UpdateLxcCacheTestCase(TestCase):
58
 
 
59
 
    def test_parse_args(self):
60
 
        args = parse_args(
61
 
            ['-d', '-v', 'user@host', 'trusty', 'ppc64el', './workspace'])
62
 
        self.assertEqual('./workspace', args.workspace)
63
 
        self.assertEqual('trusty', args.release)
64
 
        self.assertEqual('ppc64el', args.arch)
65
 
        self.assertEqual('ubuntu', args.dist)
66
 
        self.assertEqual('default', args.variant)
67
 
        self.assertTrue(args.verbose)
68
 
        self.assertTrue(args.dry_run)
69
 
 
70
 
    @patch('update_lxc_cache.LxcCache.save_index', autospec=True)
71
 
    @patch('update_lxc_cache.LxcCache.put_lxc_data', autospec=True)
72
 
    @patch('update_lxc_cache.LxcCache.get_lxc_data', autospec=True,
73
 
           return_value=('rootfs_path', 'meta_path'))
74
 
    @patch('update_lxc_cache.LxcCache.get_updates', autospec=True,
75
 
           return_value=('system', 'data'))
76
 
    def test_main(self, gu_mock, gl_mock, pl_mock, si_mock):
77
 
        lxc_cache = LxcCache('./workspace')
78
 
        rc = main(
79
 
            ['-d', '-v', 'user@host', 'trusty', 'ppc64el', './workspace'])
80
 
        self.assertEqual(0, rc)
81
 
        lxc_cache = gu_mock.call_args[0][0]
82
 
        self.assertIsInstance(lxc_cache, LxcCache)
83
 
        self.assertEqual(os.path.abspath('./workspace'), lxc_cache.workspace)
84
 
        self.assertTrue(lxc_cache.verbose)
85
 
        self.assertTrue(lxc_cache.dry_run)
86
 
        gu_mock.assert_called_with(
87
 
            lxc_cache, 'ubuntu', 'trusty', 'ppc64el', 'default')
88
 
        gl_mock.assert_called_with(lxc_cache, 'system')
89
 
        pl_mock.assert_called_with(
90
 
            lxc_cache, 'user@host', 'system', 'rootfs_path', 'meta_path')
91
 
        si_mock.assert_called_with(lxc_cache, 'data')
92
 
 
93
 
 
94
 
class LxcCacheTestCase(TestCase):
95
 
 
96
 
    def test_init(self):
97
 
        lxc_cache = LxcCache('./workspace', True, True)
98
 
        self.assertEqual(os.path.abspath('./workspace'), lxc_cache.workspace)
99
 
        self.assertTrue(lxc_cache.verbose)
100
 
        self.assertTrue(lxc_cache.dry_run)
101
 
        self.assertEqual({}, lxc_cache.systems)
102
 
 
103
 
    def test_init_with_local_cache(self):
104
 
        with temp_dir() as workspace:
105
 
            make_local_cache(workspace)
106
 
            lxc_cache = LxcCache(workspace)
107
 
        expected_systems = make_systems()
108
 
        self.assertEqual(expected_systems, lxc_cache.systems)
109
 
 
110
 
    def test_init_systems_without_local_cache(self):
111
 
        with temp_dir() as workspace:
112
 
            lxc_cache = LxcCache(workspace)
113
 
            systems, data = lxc_cache.init_systems('workspace')
114
 
        self.assertEqual({}, systems)
115
 
        self.assertIsNone(data)
116
 
 
117
 
    def test_init_systems_with_local_cache(self):
118
 
        with temp_dir() as workspace:
119
 
            index_path = make_local_cache(workspace)
120
 
            lxc_cache = LxcCache(workspace)
121
 
            systems, data = lxc_cache.init_systems(index_path)
122
 
        expected_systems = make_systems()
123
 
        self.assertEqual(expected_systems, systems)
124
 
        self.assertEqual(INDEX_DATA, data)
125
 
 
126
 
    def test_init_systems_with_remote_location(self):
127
 
        url = 'https://images.linuxcontainers.org/meta/1.0/index-system'
128
 
        response = FakeResponse(200, INDEX_DATA)
129
 
        with patch('urllib2.Request', autospec=True,
130
 
                   return_value='request') as r_mock:
131
 
            with patch('urllib2.urlopen', autospec=True,
132
 
                       return_value=response) as ul_mock:
133
 
                with temp_dir() as workspace:
134
 
                    lxc_cache = LxcCache(workspace)
135
 
                    systems, data = lxc_cache.init_systems(url)
136
 
        expected_systems = make_systems()
137
 
        self.assertEqual(expected_systems, systems)
138
 
        self.assertEqual(INDEX_DATA, data)
139
 
        ul_mock.assert_called_with('request')
140
 
        r_mock.assert_called_with(url)
141
 
 
142
 
    def test_get_updates_none(self):
143
 
        systems = make_systems()
144
 
        with temp_dir() as workspace:
145
 
            make_local_cache(workspace)
146
 
            lxc_cache = LxcCache(workspace)
147
 
            with patch.object(lxc_cache, 'init_systems', autospec=True,
148
 
                              return_value=(systems, INDEX_DATA)) as is_mock:
149
 
                new_system, new_data = lxc_cache.get_updates(
150
 
                    'ubuntu', 'trusty', 'ppc64el', 'default')
151
 
        self.assertIsNone(new_system)
152
 
        self.assertIsNone(new_data)
153
 
        is_mock.assert_called_with(
154
 
            'https://images.linuxcontainers.org/meta/1.0/index-system')
155
 
 
156
 
    def test_get_updates_found(self):
157
 
        remote_data = INDEX_DATA.replace(
158
 
            'ppc64el;default;20154', 'ppc64el;default;20159')
159
 
        remote_systems = make_systems(remote_data)
160
 
        remote_updates = (remote_systems, remote_data)
161
 
        with temp_dir() as workspace:
162
 
            make_local_cache(workspace)
163
 
            lxc_cache = LxcCache(workspace)
164
 
            with patch.object(lxc_cache, 'init_systems', autospec=True,
165
 
                              return_value=remote_updates) as is_mock:
166
 
                new_system, new_data = lxc_cache.get_updates(
167
 
                    'ubuntu', 'trusty', 'ppc64el', 'default')
168
 
        key = ('ubuntu', 'trusty', 'ppc64el', 'default')
169
 
        self.assertEqual(remote_systems[key], new_system)
170
 
        self.assertEqual(remote_data, new_data)
171
 
        is_mock.assert_called_with(
172
 
            'https://images.linuxcontainers.org/meta/1.0/index-system')
173
 
 
174
 
    def test_get_lxc_data(self):
175
 
        systems = make_systems()
176
 
        system = systems[('ubuntu', 'trusty', 'ppc64el', 'default')]
177
 
        with temp_dir() as workspace:
178
 
            make_local_cache(workspace)
179
 
            lxc_cache = LxcCache(workspace)
180
 
            with patch.object(lxc_cache, 'download', autospec=True) as d_mock:
181
 
                rootfs_path, meta_path = lxc_cache.get_lxc_data(system)
182
 
            image_path = os.path.join(
183
 
                workspace, 'images/ubuntu/trusty/ppc64el/default/20154/')
184
 
            self.assertTrue(os.path.isdir(image_path))
185
 
        expected_rootfs_path = os.path.join(image_path, 'rootfs.tar.xz')
186
 
        expected_meta_path = os.path.join(image_path, 'meta.tar.xz')
187
 
        self.assertEqual(expected_rootfs_path, rootfs_path)
188
 
        self.assertEqual(expected_meta_path, meta_path)
189
 
        d_mock.assert_any_call(
190
 
            'https://images.linuxcontainers.org'
191
 
            '/images/ubuntu/trusty/ppc64el/default/20154/rootfs.tar.xz',
192
 
            rootfs_path)
193
 
        d_mock.assert_any_call(
194
 
            'https://images.linuxcontainers.org'
195
 
            '/images/ubuntu/trusty/ppc64el/default/20154/meta.tar.xz',
196
 
            meta_path)
197
 
 
198
 
    def test_put_lxc_data(self):
199
 
        systems = make_systems()
200
 
        system = systems[('ubuntu', 'trusty', 'ppc64el', 'default')]
201
 
        with patch('subprocess.check_call', autospec=True) as cc_mock:
202
 
            lxc_cache = LxcCache('workspace')
203
 
            lxc_cache.put_lxc_data(
204
 
                'user@host', system, '/rootfs_path', '/meta_path')
205
 
        put_script = PUT_SCRIPT.format(
206
 
            user_host='user@host', rootfs_path='/rootfs_path',
207
 
            meta_path='/meta_path')
208
 
        cc_mock.assert_any_call([put_script], shell=True)
209
 
        cache_path = '/var/cache/lxc/download/ubuntu/trusty/ppc64el/default'
210
 
        install_script = INSTALL_SCRIPT.format(
211
 
            user_host='user@host', lxc_cache=cache_path,
212
 
            rootfs='rootfs.tar.xz', meta='meta.tar.xz')
213
 
        cc_mock.assert_any_call([install_script], shell=True)
214
 
 
215
 
    def test_save_index(self):
216
 
        with temp_dir() as workspace:
217
 
            lxc_cache = LxcCache(workspace)
218
 
            lxc_cache.save_index(INDEX_DATA)
219
 
            index_path = os.path.join(workspace, 'meta/1.0/index-system')
220
 
            self.assertTrue(os.path.isfile(index_path))
221
 
            with open(index_path) as f:
222
 
                data = f.read()
223
 
        self.assertEqual(INDEX_DATA, data)
224
 
 
225
 
    def test_download(self):
226
 
        response = FakeResponse(200, 'rootfs.tar.xz')
227
 
        with temp_dir() as workspace:
228
 
            with patch('urllib2.Request', autospec=True,
229
 
                       return_value='request') as r_mock:
230
 
                with patch('urllib2.urlopen', autospec=True,
231
 
                           return_value=response) as ul_mock:
232
 
                    lxc_cache = LxcCache(workspace)
233
 
                    file_path = os.path.join(workspace, 'rootfs.tar.xz')
234
 
                    lxc_cache.download('url', file_path)
235
 
            self.assertTrue(os.path.isfile(file_path))
236
 
            with open(file_path) as f:
237
 
                data = f.read()
238
 
        self.assertEqual('rootfs.tar.xz', data)
239
 
        r_mock.assert_called_with('url')
240
 
        ul_mock.assert_called_with('request')