Introduce the constrained network server.

media/tools/constrained_network_server/cns_test.py

+#!/usr/bin/python
+
+# Copyright (c) 2011 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Tests for Constrained Network Server."""
+
+import os
+import signal
+import subprocess
+import tempfile
+import time
+import unittest
+import urllib2
+
+import cns
+
+
+class PortAllocatorTest(unittest.TestCase):
+  """Unit tests for the Port Allocator class."""
+
+  # Use low expiry time for easy testing. Use 2 instead of 1 so we have an
+  # integer sleep time for staggering when / 2. Timing is too tight otherwise.
+  _EXPIRY_TIME = 2

[Ami GONE FROM CHROMIUM, 2011/11/16 03:11:22]

Ouch to real time and sleeps.
Inject a clock into PortAllocator instead?

[DaleCurtis, 2011/11/16 20:20:15]

Done.

+
+  def setUp(self):
+    # TODO(dalecurtis): Mock out actual calls to shadi's port setup.
+    self._pa = cns.PortAllocator(cns._DEFAULT_CNS_PORT_RANGE, self._EXPIRY_TIME)
+
+  def tearDown(self):
+    self._pa.Cleanup(all_ports=True)
+    # Ensure ports are cleaned properly.
+    self.assertEquals(self._pa._ports, {})
+
+  def testPortAllocator(self):
+    # Ensure Get() succeeds and returns the correct port.
+    self.assertEquals(self._pa.Get('test'), cns._DEFAULT_CNS_PORT_RANGE[0])
+
+    # Call again with the same key and make sure we get the same port.
+    self.assertEquals(self._pa.Get('test'), cns._DEFAULT_CNS_PORT_RANGE[0])
+
+    # Call with a different key and make sure we get a different port.
+    self.assertEquals(self._pa.Get('test2'), cns._DEFAULT_CNS_PORT_RANGE[0] + 1)
+
+    # Wait for ports to expire.
+    time.sleep(self._EXPIRY_TIME)
+
+    # Test to make sure cache is checked before expiring ports.
+    self.assertEquals(self._pa.Get('test2'), cns._DEFAULT_CNS_PORT_RANGE[0] + 1)
+
+    # Wait for ports to expire again.
+    time.sleep(self._EXPIRY_TIME)
+
+    # Request a new port, old ports should be expired, so we should get the
+    # first port in the range. Make sure this is the only allocated port.
+    self.assertEquals(self._pa.Get('test3'), cns._DEFAULT_CNS_PORT_RANGE[0])
+    self.assertEquals(self._pa._ports.keys(), [cns._DEFAULT_CNS_PORT_RANGE[0]])
+
+  def testPortAllocatorExpiresOnlyCorrectPorts(self):
+    # Ensure Get() succeeds and returns the correct port.
+    self.assertEquals(self._pa.Get('test'), cns._DEFAULT_CNS_PORT_RANGE[0])
+
+    # Stagger port allocation and so we can ensure only ports older than the
+    # expiry time are actually expired.
+    time.sleep(self._EXPIRY_TIME / 2)
+
+    # Call with a different key and make sure we get a different port.
+    self.assertEquals(self._pa.Get('test2'), cns._DEFAULT_CNS_PORT_RANGE[0] + 1)
+
+    # After this sleep the port with key 'test' should expire on the next Get().
+    time.sleep(self._EXPIRY_TIME / 2)
+
+    # Call with a different key and make sure we get the first port.
+    self.assertEquals(self._pa.Get('test3'), cns._DEFAULT_CNS_PORT_RANGE[0])
+    self.assertEquals(set(self._pa._ports.keys()), set([
+        cns._DEFAULT_CNS_PORT_RANGE[0], cns._DEFAULT_CNS_PORT_RANGE[0] + 1]))
+
+
+class ConstrainedNetworkServerTest(unittest.TestCase):
+  """End to end tests for ConstrainedNetworkServer system."""
+
+  # Amount of time to wait for the CNS to start up.
+  _SERVER_START_SLEEP_SECS = 1
+
+  # Sample data used to verify file serving.
+  _TEST_DATA = 'The quick brown fox jumps over the lazy dog'
+
+  # Server information.
+  _SERVER_PORT = 8000
+  _SERVER_URL = 'http://127.0.0.1:%d/ServeConstrained?' % _SERVER_PORT
+
+  # Setting for latency testing.
+  _LATENCY_TEST_SECS = 5
+
+  def _StartServer(self):
+    """Starts the CNS, returns pid."""
+    cmd = ['python', 'cns.py', '--port', str(self._SERVER_PORT)]
+    process = subprocess.Popen(cmd)
+
+    # Wait for server to startup.
+    time.sleep(self._SERVER_START_SLEEP_SECS)

[Ami GONE FROM CHROMIUM, 2011/11/16 03:11:22]

Ouch.  Can you instead emit a stdout from cns saying server is up when it's
ready and wait for that to show up in subprocess here?  

sleep==donotwant

[DaleCurtis, 2011/11/16 20:20:15]

Done.

+    return process.pid
+
+  def setUp(self):
+    # Create temp file for serving.
+    f, self._file = tempfile.mkstemp(dir=os.getcwd())
+    os.write(f, self._TEST_DATA)
+    os.close(f)
+
+    # Strip cwd off so we have a proper relative path.
+    self._relative_fn = self._file[len(os.getcwd()) + 1:]
+
+    # Start the CNS.
+    self._server_pid = self._StartServer()
+
+  def tearDown(self):
+    os.unlink(self._file)
+    os.kill(self._server_pid, signal.SIGKILL)
+
+  def testServerServesFiles(self):
+    f = urllib2.urlopen('%sf=%s' % (self._SERVER_URL, self._relative_fn))
+
+    # Verify file data is served correctly.
+    self.assertEqual(self._TEST_DATA, f.read())
+
+  def testServerLatencyConstraint(self):
+    now = time.time()
+
+    base_url = '%sf=%s' % (self._SERVER_URL, self._relative_fn)
+    url = '%s&latency=%d' % (base_url, self._LATENCY_TEST_SECS * 1000)
+    f = urllib2.urlopen(url)
+
+    # Verify file data is served correctly.
+    self.assertEqual(self._TEST_DATA, f.read())
+
+    # Verify the request took longer than the requested latency.

[Ami GONE FROM CHROMIUM, 2011/11/16 03:11:22]

Would be a more convincing test if the un-latency'd test asserted it took *less*
than _L_T_S :)

[DaleCurtis, 2011/11/16 20:20:15]

Done.

+    self.assertTrue(time.time() - now > self._LATENCY_TEST_SECS)
+
+    # Verify the server properly redirected the URL.
+    self.assertEquals(f.geturl(), base_url.replace(
+        str(self._SERVER_PORT), str(cns._DEFAULT_CNS_PORT_RANGE[0])))
+
+
+if __name__ == '__main__':
+  unittest.main()