dev_server_unittest.py revision 761341eafc11de617743ca8126a3b1a3b9f9c430 
1#!/usr/bin/python
2#
3# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6
7"""Unit tests for client/common_lib/cros/dev_server.py."""
8
9import httplib
10import logging
11import mox
12import StringIO
13import unittest
14import urllib2
15
16from autotest_lib.client.common_lib.cros import dev_server
17
18
19class DevServerTest(mox.MoxTestBase):
20    """Unit tests for dev_server.DevServer.
21
22    @var _HOST: fake dev server host address.
23    """
24
25    _HOST = 'http://nothing'
26    _CRASH_HOST = 'http://nothing-crashed'
27
28    def setUp(self):
29        super(DevServerTest, self).setUp()
30        self.dev_server = dev_server.DevServer(self._HOST, self._CRASH_HOST)
31
32
33    def _returnHttpServerError(self):
34        self.mox.StubOutWithMock(urllib2, 'urlopen')
35        e500 = urllib2.HTTPError(url='',
36                                 code=httplib.INTERNAL_SERVER_ERROR,
37                                 msg='',
38                                 hdrs=None,
39                                 fp=StringIO.StringIO('Expected.'))
40        urllib2.urlopen(mox.IgnoreArg()).AndRaise(e500)
41
42
43    def _returnHttpForbidden(self):
44        self.mox.StubOutWithMock(urllib2, 'urlopen')
45        e403 = urllib2.HTTPError(url='',
46                                 code=httplib.FORBIDDEN,
47                                 msg='',
48                                 hdrs=None,
49                                 fp=StringIO.StringIO('Expected.'))
50        urllib2.urlopen(mox.IgnoreArg()).AndRaise(e403)
51
52
53    def testSuccessfulTriggerDownloadSync(self):
54        """Call the dev server's download method with synchronous=True."""
55        name = 'fake/image'
56        self.mox.StubOutWithMock(urllib2, 'urlopen')
57        self.mox.StubOutWithMock(dev_server.DevServer, 'finish_download')
58        to_return = StringIO.StringIO('Success')
59        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
60                                mox.StrContains(name))).AndReturn(to_return)
61        self.dev_server.finish_download(name)
62
63        # Synchronous case requires a call to finish download.
64        self.mox.ReplayAll()
65        self.dev_server.trigger_download(name, synchronous=True)
66        self.mox.VerifyAll()
67
68
69    def testSuccessfulTriggerDownloadASync(self):
70        """Call the dev server's download method with synchronous=False."""
71        name = 'fake/image'
72        self.mox.StubOutWithMock(urllib2, 'urlopen')
73        self.mox.StubOutWithMock(dev_server.DevServer, 'finish_download')
74        to_return = StringIO.StringIO('Success')
75        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
76                                mox.StrContains(name))).AndReturn(to_return)
77
78        self.mox.ReplayAll()
79        self.dev_server.trigger_download(name, synchronous=False)
80        self.mox.VerifyAll()
81
82
83    def testErrorTriggerDownload(self):
84        """Should call the dev server's download method, fail gracefully."""
85        self._returnHttpForbidden()
86        self.mox.ReplayAll()
87        self.assertRaises(dev_server.DevServerException,
88                          self.dev_server.trigger_download,
89                          '')
90
91
92    def testForbiddenTriggerDownload(self):
93        """Should call the dev server's download method, get exception."""
94        self._returnHttpForbidden()
95        self.mox.ReplayAll()
96        self.assertRaises(dev_server.DevServerException,
97                          self.dev_server.trigger_download,
98                          '')
99
100
101    def testSuccessfulFinishDownload(self):
102        """Should successfully call the dev server's finish download method."""
103        name = 'fake/image'
104        self.mox.StubOutWithMock(urllib2, 'urlopen')
105        to_return = StringIO.StringIO('Success')
106        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
107                                mox.StrContains(name))).AndReturn(to_return)
108
109        # Synchronous case requires a call to finish download.
110        self.mox.ReplayAll()
111        self.dev_server.finish_download(name)  # Raises on failure.
112        self.mox.VerifyAll()
113
114
115    def testErrorTriggerDownload(self):
116        """Should call the dev server's finish download method, fail gracefully.
117        """
118        self._returnHttpServerError()
119        self.mox.ReplayAll()
120        self.assertRaises(dev_server.DevServerException,
121                          self.dev_server.finish_download,
122                          '')
123
124
125    def testListControlFiles(self):
126        """Should successfully list control files from the dev server."""
127        name = 'fake/build'
128        control_files = ['file/one', 'file/two']
129        self.mox.StubOutWithMock(urllib2, 'urlopen')
130        to_return = StringIO.StringIO('\n'.join(control_files))
131        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
132                                mox.StrContains(name))).AndReturn(to_return)
133        self.mox.ReplayAll()
134        paths = self.dev_server.list_control_files(name)
135        self.assertEquals(len(paths), 2)
136        for f in control_files:
137            self.assertTrue(f in paths)
138
139
140    def testFailedListControlFiles(self):
141        """Should call the dev server's list-files method, get exception."""
142        self._returnHttpServerError()
143        self.mox.ReplayAll()
144        self.assertRaises(dev_server.DevServerException,
145                          self.dev_server.list_control_files,
146                          '')
147
148
149    def testExplodingListControlFiles(self):
150        """Should call the dev server's list-files method, get exception."""
151        self._returnHttpForbidden()
152        self.mox.ReplayAll()
153        self.assertRaises(dev_server.DevServerException,
154                          self.dev_server.list_control_files,
155                          '')
156
157
158    def testGetControlFile(self):
159        """Should successfully get a control file from the dev server."""
160        name = 'fake/build'
161        file = 'file/one'
162        contents = 'Multi-line\nControl File Contents\n'
163        self.mox.StubOutWithMock(urllib2, 'urlopen')
164        to_return = StringIO.StringIO(contents)
165        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
166                                mox.StrContains(name),
167                                mox.StrContains(file))).AndReturn(to_return)
168        self.mox.ReplayAll()
169        self.assertEquals(self.dev_server.get_control_file(name, file),
170                          contents)
171
172
173    def testErrorGetControlFile(self):
174        """Should try to get the contents of a control file, get exception."""
175        self._returnHttpServerError()
176        self.mox.ReplayAll()
177        self.assertRaises(dev_server.DevServerException,
178                          self.dev_server.get_control_file,
179                          '', '')
180
181
182    def testForbiddenGetControlFile(self):
183        """Should try to get the contents of a control file, get exception."""
184        self._returnHttpForbidden()
185        self.mox.ReplayAll()
186        self.assertRaises(dev_server.DevServerException,
187                          self.dev_server.get_control_file,
188                          '', '')
189
190
191    def testGetLatestBuild(self):
192        """Should successfully return a build for a given target."""
193        target = 'x86-generic-release'
194        build_string = 'R18-1586.0.0-a1-b1514'
195        self.mox.StubOutWithMock(urllib2, 'urlopen')
196        to_return = StringIO.StringIO(build_string)
197        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
198                                mox.StrContains(target))).AndReturn(to_return)
199        self.mox.ReplayAll()
200        build = self.dev_server.get_latest_build(target)
201        self.assertEquals(build_string, build)
202
203
204    def testThatWeCorrectlyReHashToTheSameDevserver(self):
205        """Ensure calls with same hashing_value go to the same devserver."""
206        for index in range(10):
207          self.dev_server._dev_servers.append('http://nothing_%d' % index)
208
209        method_name = 'my_method'
210        hv1 = 'iliketacos'
211        hv2 = 'iliketacos'
212        hv3 = 'idontliketacos :('
213        hv4 = 'idontliketacos :('
214
215        call1 = self.dev_server._build_call(method_name, hashing_value=hv1)
216        call2 = self.dev_server._build_call(method_name, hashing_value=hv2,
217                                            some_arg='value')
218        call3 = self.dev_server._build_call(method_name, hashing_value=hv3)
219        call4 = self.dev_server._build_call(method_name, hashing_value=hv4,
220                                            some_arg='value')
221
222        self.assertTrue(call2.startswith(call1))
223        self.assertTrue(call4.startswith(call3))
224
225
226    def testGetLatestBuildWithManyDevservers(self):
227        """Should successfully return newest build with multiple devservers."""
228        self.dev_server._dev_servers.append('http://nothing_2')
229        self.dev_server._dev_servers.append('http://nothing_3')
230        target = 'x86-generic-release'
231        build_string1 = 'R9-1586.0.0-a1-b1514'
232        build_string2 = 'R19-1586.0.0-a1-b3514'
233        build_string3 = 'R18-1486.0.0-a1-b2514'
234        self.mox.StubOutWithMock(urllib2, 'urlopen')
235        to_return1 = StringIO.StringIO(build_string1)
236        to_return2 = StringIO.StringIO(build_string2)
237        to_return3 = StringIO.StringIO(build_string3)
238        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
239                                mox.StrContains(target))).AndReturn(to_return1)
240        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
241                                mox.StrContains(target))).AndReturn(to_return2)
242        urllib2.urlopen(mox.And(mox.StrContains(self._HOST),
243                                mox.StrContains(target))).AndReturn(to_return3)
244
245        self.mox.ReplayAll()
246        build = self.dev_server.get_latest_build(target)
247        self.assertEquals(build_string2, build)
248
249
250    def testCrashesAreSetToTheCrashServer(self):
251        """Should send symbolicate dump rpc calls to crash_server."""
252        hv = 'iliketacos'
253        self.mox.ReplayAll()
254        call = self.dev_server._build_call('symbolicate_dump', hashing_value=hv)
255        self.assertTrue(call.startswith(self._CRASH_HOST))
256