Implement chrome.experimental.downloads.onChanged

net/tools/testserver/testserver.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.
 
 """This is a simple HTTP/FTP/SYNC/TCP/UDP/ server used for testing Chrome.
 
 It supports several test URLs, as specified by the handlers in TestPageHandler.
 By default, it listens on an ephemeral port and sends the port number back to
 the originating process over a pipe. The originating process can specify an
@@ skipping 890 matching lines @@
     self.wfile.write(data)
 
     return True
 
   def FileHandler(self):
     """This handler sends the contents of the requested file.  Wow, it's like
     a real webserver!"""
     prefix = self.server.file_root_url
     if not self.path.startswith(prefix):
       return False
-    # Consume a request body if present.
-    if self.command == 'POST' or self.command == 'PUT' :
-      self.ReadRequestBody()
     return self._FileHandlerHelper(prefix)
 
   def PostOnlyFileHandler(self):
     """This handler sends the contents of the requested file on a POST."""
     prefix = urlparse.urljoin(self.server.file_root_url, 'post/')
     if not self.path.startswith(prefix):
       return False
-    self.ReadRequestBody()
     return self._FileHandlerHelper(prefix)
 
   def _FileHandlerHelper(self, prefix):
+    request_body = ''
+    if self.command == 'POST' or self.command == 'PUT':
+      # Consume a request body if present.
+      request_body = self.ReadRequestBody()
+
     _, _, url_path, _, query, _ = urlparse.urlparse(self.path)
+    query_dict = cgi.parse_qs(query)
+
+    expected_body = query_dict.get('expected_body', [])

[cbentzel, 2012/02/14 00:58:46]

Should the expected_body argument be base64 encoded? Perhaps too much effort
now.

[benjhayden, 2012/02/14 15:52:11]

If somebody has a need for it, they can add it.

+    if expected_body and request_body not in expected_body:

[cbentzel, 2012/02/13 21:58:01]

if request_body not in expected_body: should be sufficient.

[benjhayden, 2012/02/13 22:05:51]

There are tests that do not specify expected_body, so expected_body will be [].
No possible request_body is in [], so this code would return 404 to those tests
instead of succeeding.

[cbentzel, 2012/02/14 00:58:46]

Yeah, my mistake. 

You could do

  expected_body = query_dict.get('expected_body', [request_body])

but that would probably be less clear than this.

[benjhayden, 2012/02/14 15:52:11]

It doesn't sound like there's a strong preference, so I'll leave it how it is.

+      self.send_response(404)
+      self.end_headers()
+      self.wfile.write('')
+      return True
+
+    expected_headers = query_dict.get('expected_headers', [])
+    for expected_header in expected_headers:
+      header_name, expected_value = expected_header.split(':')
+      if self.headers.getheader(header_name) != expected_value:
+        self.send_response(404)
+        self.end_headers()
+        self.wfile.write('')
+        return True
+
     sub_path = url_path[len(prefix):]
     entries = sub_path.split('/')
     file_path = os.path.join(self.server.data_dir, *entries)
     if os.path.isdir(file_path):
       file_path = os.path.join(file_path, 'index.html')
 
     if not os.path.isfile(file_path):
       print "File not found " + sub_path + " full path:" + file_path
       self.send_error(404)
       return True
@@ skipping 1069 matching lines @@
                            'random key if none is specified on the command '
                            'line.')
   option_parser.add_option('', '--policy-user', default='user@example.com',
                            dest='policy_user',
                            help='Specify the user name the server should '
                            'report back to the client as the user owning the '
                            'token used for making the policy request.')
   options, args = option_parser.parse_args()
 
   sys.exit(main(options, args))