Index: src/platform/dev/update_test.py |
diff --git a/src/platform/dev/update_test.py b/src/platform/dev/update_test.py |
new file mode 100755 |
index 0000000000000000000000000000000000000000..7c726948c6c033cfd5bf6910d4b89b13d8ee9d77 |
--- /dev/null |
+++ b/src/platform/dev/update_test.py |
@@ -0,0 +1,105 @@ |
+#!/usr/bin/python |
+ |
+import urllib2 |
+import sys |
+import threading |
+from urlparse import urljoin |
+from xml.dom import minidom |
+ |
+UPDATE_BLOB="""\ |
+<?xml version="1.0" encoding="UTF-8"?> |
+<o:gupdate |
+ xmlns:o="http://www.google.com/update2/request" |
+ version="MementoSoftwareUpdate-0.1.0.0" |
+ protocol="2.0" |
+ machineid="{1B0A13AC-7004-638C-3CA6-EC082E8F5DE9}" |
+ ismachine="0" |
+ userid="{bogus}"> |
+<o:os version="Memento" |
+ platform="memento" |
+ sp="ForcedUpdate_i686"> |
+</o:os> |
+<o:app appid="{87efface-864d-49a5-9bb3-4b050a7c227a}" |
+ version="ForcedUpdate" |
+ lang="en-us" |
+ brand="GGLG" |
+ track="developer-build" |
+ board="x86-generic"> |
+<o:ping active="0"></o:ping> |
+<o:updatecheck></o:updatecheck> |
+</o:app> |
+</o:gupdate> |
+""" |
+ |
+UPDATE_URL = 'http://localhost:8080/update/' |
+ |
+def do_ping(): |
+ update_ping = urllib2.Request(UPDATE_URL, update_blob) |
+ update_ping.add_header('Content-Type', 'text/xml') |
+ print urllib2.urlopen(update_ping).read() |
+ #TODO assert noupdate |
+ |
+def do_version_ping(): |
+ url = UPDATE_URL + 'LATEST' |
+ update_ping = urllib2.Request(url, UPDATE_BLOB) |
+ update_ping.add_header('Content-Type', 'text/xml') |
+ response = urllib2.urlopen(update_ping).read() |
+ assert _verify_response(response), 'couldn\'t fetch update file' |
+ print 'Successfully pinged updateserver.' |
+ |
+def do_badversion_ping(): |
+ url = UPDATE_URL + 'BADVERSION' |
+ update_ping = urllib2.Request(url, UPDATE_BLOB) |
+ update_ping.add_header('Content-Type', 'text/xml') |
+ response = urllib2.urlopen(update_ping).read() |
+ assert ('noupdate' in response) |
+ print 'requesting bogus version returns noupdate.' |
+ |
+def _verify_download(url, content_length): |
+ # Eventually, verify something about the update. For now, |
+ # sanity-check its size. |
+ f = urllib2.urlopen(url) |
+ data = f.read(1024 * 1024) |
+ length = len(data) |
+ while data: |
+ data = f.read(1024 * 1024) |
+ length += len(data) |
+ assert content_length == length |
+ print 'Got a valid download.' |
+ return True |
+ |
+def _verify_response(data): |
+ update_dom = minidom.parseString(data) |
+ print data |
+ root = update_dom.firstChild |
+ update_info = root.getElementsByTagName('updatecheck')[0] |
+ update_url = update_info.getAttribute('codebase') |
+ hash = update_info.getAttribute('hash') |
+ head_request = urllib2.Request(update_url) |
+ head_request.get_method = lambda: 'HEAD' |
+ try: |
+ fd = urllib2.urlopen(head_request) |
+ except urllib2.HTTPError, e: |
+ print 'FAILED: unable to retrieve %s\n\t%s' % (update_url, e) |
+ return False |
+ length = int(fd.headers.getheaders('Content-Length')[0]) |
+ assert length > 0 |
+ print 'Got a valid update response.' |
+ fd.close() |
+ assert (urllib2.urlopen(urljoin(update_url, 'cksum')).read() == hash) |
+ print 'Update cksum matched the one in the update XML.' |
+ return _verify_download(update_url, length) |
+ |
+def test(num_clients): |
+ # Fake some concurrent requests for each autoupdate operation. |
+ for clients in range(num_clients): |
+ for op in (do_version_ping, do_badversion_ping): |
+ t = threading.Thread(target=op) |
+ t.start() |
+ |
+if __name__ == '__main__': |
+ if len(sys.argv) > 1: |
+ num_clients = int(sys.argv[1]) |
+ else: |
+ num_clients = 1 |
+ test(num_clients) |