OLD | NEW |
---|---|
1 #!/usr/bin/env python | 1 #!/usr/bin/env python |
2 # Copyright (c) 2012 The Chromium Authors. All rights reserved. | 2 # Copyright (c) 2012 The Chromium Authors. All rights reserved. |
3 # Use of this source code is governed by a BSD-style license that can be | 3 # Use of this source code is governed by a BSD-style license that can be |
4 # found in the LICENSE file. | 4 # found in the LICENSE file. |
5 | 5 |
6 """Tests for Constrained Network Server.""" | 6 """Tests for Constrained Network Server.""" |
7 import os | 7 import os |
8 import signal | 8 import signal |
9 import subprocess | 9 import subprocess |
10 import tempfile | 10 import tempfile |
(...skipping 18 matching lines...) Expand all Loading... | |
29 # Mock out time.time() to accelerate port expiration testing. | 29 # Mock out time.time() to accelerate port expiration testing. |
30 self._old_time = time.time | 30 self._old_time = time.time |
31 self._current_time = 0 | 31 self._current_time = 0 |
32 time.time = lambda: self._current_time | 32 time.time = lambda: self._current_time |
33 | 33 |
34 # TODO(dalecurtis): Mock out actual calls to shadi's port setup. | 34 # TODO(dalecurtis): Mock out actual calls to shadi's port setup. |
35 self._pa = cns.PortAllocator(cns._DEFAULT_CNS_PORT_RANGE, self._EXPIRY_TIME) | 35 self._pa = cns.PortAllocator(cns._DEFAULT_CNS_PORT_RANGE, self._EXPIRY_TIME) |
36 self._MockTrafficControl() | 36 self._MockTrafficControl() |
37 | 37 |
38 def tearDown(self): | 38 def tearDown(self): |
39 self._pa.Cleanup(_INTERFACE, all_ports=True) | 39 self._pa.Cleanup(all_ports=True) |
DaleCurtis
2012/08/07 23:53:57
Is this no longer necessary / incorrect?
shadi
2012/08/08 22:55:49
Not necessary, Cleanup() does a clean up on alloca
| |
40 # Ensure ports are cleaned properly. | 40 # Ensure ports are cleaned properly. |
41 self.assertEquals(self._pa._ports, {}) | 41 self.assertEquals(self._pa._ports, {}) |
42 time.time = self._old_time | 42 time.time = self._old_time |
43 self._RestoreTrafficControl() | 43 self._RestoreTrafficControl() |
44 | 44 |
45 def _MockTrafficControl(self): | 45 def _MockTrafficControl(self): |
46 self.old_CreateConstrainedPort = traffic_control.CreateConstrainedPort | 46 self.old_CreateConstrainedPort = traffic_control.CreateConstrainedPort |
47 self.old_DeleteConstrainedPort = traffic_control.DeleteConstrainedPort | 47 self.old_DeleteConstrainedPort = traffic_control.DeleteConstrainedPort |
48 self.old_TearDown = traffic_control.TearDown | 48 self.old_TearDown = traffic_control.TearDown |
49 | 49 |
(...skipping 60 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
110 self._current_time += self._EXPIRY_TIME | 110 self._current_time += self._EXPIRY_TIME |
111 | 111 |
112 # Send second Get() which would normally cause ports to expire. Ensure that | 112 # Send second Get() which would normally cause ports to expire. Ensure that |
113 # the ports did not expire. | 113 # the ports did not expire. |
114 self.assertEquals(self._pa.Get('test2'), cns._DEFAULT_CNS_PORT_RANGE[0] + 1) | 114 self.assertEquals(self._pa.Get('test2'), cns._DEFAULT_CNS_PORT_RANGE[0] + 1) |
115 self.assertEquals(set(self._pa._ports.keys()), set([ | 115 self.assertEquals(set(self._pa._ports.keys()), set([ |
116 cns._DEFAULT_CNS_PORT_RANGE[0], cns._DEFAULT_CNS_PORT_RANGE[0] + 1])) | 116 cns._DEFAULT_CNS_PORT_RANGE[0], cns._DEFAULT_CNS_PORT_RANGE[0] + 1])) |
117 | 117 |
118 | 118 |
119 class ConstrainedNetworkServerTest(unittest.TestCase): | 119 class ConstrainedNetworkServerTest(unittest.TestCase): |
120 """End to end tests for ConstrainedNetworkServer system.""" | 120 """End to end tests for ConstrainedNetworkServer system. |
121 | |
122 These tests require root access and run the cherrypy server along with | |
123 tc/iptables commands. | |
124 """ | |
121 | 125 |
122 # Amount of time to wait for the CNS to start up. | 126 # Amount of time to wait for the CNS to start up. |
123 _SERVER_START_SLEEP_SECS = 1 | 127 _SERVER_START_SLEEP_SECS = 1 |
124 | 128 |
125 # Sample data used to verify file serving. | 129 # Sample data used to verify file serving. |
126 _TEST_DATA = 'The quick brown fox jumps over the lazy dog' | 130 _TEST_DATA = 'The quick brown fox jumps over the lazy dog' |
127 | 131 |
128 # Server information. | 132 # Server information. |
129 _SERVER_URL = ('http://127.0.0.1:%d/ServeConstrained?' % | 133 _SERVER_URL = ('http://127.0.0.1:%d/ServeConstrained?' % |
130 cns._DEFAULT_SERVING_PORT) | 134 cns._DEFAULT_SERVING_PORT) |
(...skipping 54 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
185 url = '%s&latency=%d' % (base_url, self._LATENCY_TEST_SECS * 1000) | 189 url = '%s&latency=%d' % (base_url, self._LATENCY_TEST_SECS * 1000) |
186 f = urllib2.urlopen(url) | 190 f = urllib2.urlopen(url) |
187 | 191 |
188 # Verify file data is served correctly. | 192 # Verify file data is served correctly. |
189 self.assertEqual(self._TEST_DATA, f.read()) | 193 self.assertEqual(self._TEST_DATA, f.read()) |
190 | 194 |
191 # Verify the request took longer than the requested latency. | 195 # Verify the request took longer than the requested latency. |
192 self.assertTrue(time.time() - now > self._LATENCY_TEST_SECS) | 196 self.assertTrue(time.time() - now > self._LATENCY_TEST_SECS) |
193 | 197 |
194 # Verify the server properly redirected the URL. | 198 # Verify the server properly redirected the URL. |
195 self.assertEquals(f.geturl(), base_url.replace( | 199 self.assertTrue(f.geturl().startswith(base_url.replace( |
196 str(cns._DEFAULT_SERVING_PORT), str(cns._DEFAULT_CNS_PORT_RANGE[0]))) | 200 str(cns._DEFAULT_SERVING_PORT), str(cns._DEFAULT_CNS_PORT_RANGE[0])))) |
197 | 201 |
198 | 202 |
199 if __name__ == '__main__': | 203 if __name__ == '__main__': |
200 unittest.main() | 204 unittest.main() |
OLD | NEW |