Enable CNS to serve files from different port.

media/tools/constrained_network_server/cns_test.py

 #!/usr/bin/env python
 # Copyright (c) 2012 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 cherrypy
 import cns
 import traffic_control
 
 # The local interface to test on.
 _INTERFACE = 'lo'
 
 
 class PortAllocatorTest(unittest.TestCase):
   """Unit tests for the Port Allocator class."""
 
   # Expiration time for ports. In mock time.
   _EXPIRY_TIME = 6
 
   def setUp(self):
     # Mock out time.time() to accelerate port expiration testing.
     self._old_time = time.time
     self._current_time = 0
     time.time = lambda: self._current_time
 
     # TODO(dalecurtis): Mock out actual calls to shadi's port setup.
     self._pa = cns.PortAllocator(cns._DEFAULT_CNS_PORT_RANGE, self._EXPIRY_TIME)
     self._MockTrafficControl()
 
   def tearDown(self):
-    self._pa.Cleanup(_INTERFACE, all_ports=True)
+    self._pa.Cleanup(all_ports=True)
     # Ensure ports are cleaned properly.
     self.assertEquals(self._pa._ports, {})
     time.time = self._old_time
     self._RestoreTrafficControl()
 
   def _MockTrafficControl(self):
     self.old_CreateConstrainedPort = traffic_control.CreateConstrainedPort
     self.old_DeleteConstrainedPort = traffic_control.DeleteConstrainedPort
     self.old_TearDown = traffic_control.TearDown
 
@@ skipping 60 matching lines @@
     self._current_time += self._EXPIRY_TIME
 
     # Send second Get() which would normally cause ports to expire. Ensure that
     # the ports did not expire.
     self.assertEquals(self._pa.Get('test2'), cns._DEFAULT_CNS_PORT_RANGE[0] + 1)
     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."""
+  """End to end tests for ConstrainedNetworkServer system.
+
+  These tests require root access and run the cherrypy server along with
+  tc/iptables commands.
+  """
 
   # 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_URL = ('http://127.0.0.1:%d/ServeConstrained?' %
                  cns._DEFAULT_SERVING_PORT)
@@ skipping 54 matching lines @@
     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.
     self.assertTrue(time.time() - now > self._LATENCY_TEST_SECS)
 
     # Verify the server properly redirected the URL.
-    self.assertEquals(f.geturl(), base_url.replace(
-        str(cns._DEFAULT_SERVING_PORT), str(cns._DEFAULT_CNS_PORT_RANGE[0])))
+    self.assertTrue(f.geturl().startswith(base_url.replace(
+        str(cns._DEFAULT_SERVING_PORT), str(cns._DEFAULT_CNS_PORT_RANGE[0]))))
+
+
+class ConstrainedNetworkServerUnitTests(unittest.TestCase):
+  """ConstrainedNetworkServer class unit tests."""
+
+  def testGetServerURL(self):
+    """Test server URL is correct when using Cherrypy port."""
+    cns_obj = cns.ConstrainedNetworkServer(self.DummyOptions(), None)
+
+    self.assertEqual(cns_obj._GetServerURL('ab/xz.webm', port=1234, t=1),
+                     'http://127.0.0.1:1234/ServeConstrained?f=ab/xz.webm&t=1')
+
+  def testGetServerURLWithLocalServer(self):
+    """Test server URL is correct when using --local-server-port port."""
+    cns_obj = cns.ConstrainedNetworkServer(self.DummyOptionsWithServer(), None)
+
+    self.assertEqual(cns_obj._GetServerURL('ab/xz.webm', port=1234, t=1),
+                     'http://127.0.0.1:1234/media/ab/xz.webm?t=1')
+
+  class DummyOptions(object):
+    www_root = 'media'
+    port = 9000
+    cherrypy.url = lambda: 'http://127.0.0.1:9000/ServeConstrained'
+    local_server_port = None
+
+  class DummyOptionsWithServer(object):
+    www_root = 'media'
+    port = 9000
+    cherrypy.url = lambda: 'http://127.0.0.1:9000/ServeConstrained'
+    local_server_port = 8080
 
 
 if __name__ == '__main__':
   unittest.main()