| #!/usr/bin/python |
| |
| # Copyright (c) 2010 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Regression tests for devserver.""" |
| |
| import json |
| from xml.dom import minidom |
| import os |
| import psutil |
| import shutil |
| import signal |
| import subprocess |
| import tempfile |
| import time |
| import unittest |
| import urllib2 |
| |
| |
| # Paths are relative to this script's base directory. |
| LABEL = 'devserver' |
| TEST_IMAGE_PATH = 'testdata/devserver' |
| TEST_IMAGE_NAME = 'update.gz' |
| TEST_IMAGE = TEST_IMAGE_PATH + '/' + TEST_IMAGE_NAME |
| EXPECTED_HASH = 'kGcOinJ0vA8vdYX53FN0F5BdwfY=' |
| |
| # Update request based on Omaha v2 protocol format. |
| UPDATE_REQUEST = """<?xml version="1.0" encoding="UTF-8"?> |
| <request version="ChromeOSUpdateEngine-0.1.0.0" updaterversion="ChromeOSUpdateEngine-0.1.0.0" protocol="3.0" ismachine="1"> |
| <os version="Indy" platform="Chrome OS" sp="0.11.254.2011_03_09_1814_i686"></os> |
| <app appid="{DEV-BUILD}" version="11.254.2011_03_09_1814" lang="en-US" track="developer-build" board="x86-generic" hardware_class="BETA DVT" delta_okay="true"> |
| <updatecheck></updatecheck> |
| <event eventtype="3" eventresult="2" previousversion="0.11.216.2011_03_02_1358"></event> |
| </app> |
| </request> |
| """ |
| |
| # TODO(girts): use a random available port. |
| UPDATE_URL = 'http://127.0.0.1:8080/update/devserver' |
| CHECK_HEALTH_URL = 'http://127.0.0.1:8080/check_health' |
| STATIC_URL = 'http://127.0.0.1:8080/static/' |
| SERVE_URL = STATIC_URL + LABEL + '/' |
| |
| API_HOST_INFO_BAD_URL = 'http://127.0.0.1:8080/api/hostinfo/' |
| API_HOST_INFO_URL = API_HOST_INFO_BAD_URL + '127.0.0.1' |
| |
| API_SET_UPDATE_BAD_URL = 'http://127.0.0.1:8080/api/setnextupdate/' |
| API_SET_UPDATE_URL = API_SET_UPDATE_BAD_URL + '127.0.0.1' |
| |
| API_SET_UPDATE_REQUEST = 'new_update-test/the-new-update' |
| |
| DEVSERVER_START_TIMEOUT = 15 |
| |
| class DevserverTest(unittest.TestCase): |
| """Regressions tests for devserver.""" |
| |
| def setUp(self): |
| """Copies in testing files.""" |
| |
| # Copy in developer-test.gz, as "static/" directory is hardcoded, and it |
| # would be very hard to change it (static file serving is handled deep |
| # inside webpy). |
| self.test_data_path = tempfile.mkdtemp() |
| self.src_dir = os.path.dirname(__file__) |
| |
| # Copy the payload to the root of static_dir. |
| self.image_src = os.path.join(self.src_dir, TEST_IMAGE) |
| self.image = os.path.join(self.test_data_path, TEST_IMAGE_NAME) |
| shutil.copy(self.image_src, self.image) |
| |
| # Copy the payload to the location of the update label "devserver." |
| os.makedirs(os.path.join(self.test_data_path, LABEL)) |
| shutil.copy(self.image_src, os.path.join(self.test_data_path, LABEL, |
| TEST_IMAGE_NAME)) |
| |
| # Copy the payload to the location of forced label. |
| os.makedirs(os.path.join(self.test_data_path, API_SET_UPDATE_REQUEST)) |
| shutil.copy(self.image_src, os.path.join(self.test_data_path, |
| API_SET_UPDATE_REQUEST, |
| TEST_IMAGE_NAME)) |
| |
| self.pidfile = tempfile.mktemp('devserver_unittest') |
| self.devserver_process = self._StartServer(self.pidfile) |
| |
| def tearDown(self): |
| """Removes testing files.""" |
| shutil.rmtree(self.test_data_path) |
| os.kill(self.devserver_process.pid, signal.SIGKILL) |
| |
| # Helper methods begin here. |
| |
| def _StartServer(self, pidfile): |
| """Starts devserver, returns process.""" |
| cmd = [ |
| 'python', |
| os.path.join(self.src_dir, 'devserver.py'), |
| 'devserver.py', |
| '--static_dir', self.test_data_path, |
| '--pidfile', pidfile |
| ] |
| if pidfile: |
| cmd.extend(['--pidfile', pidfile]) |
| |
| process = subprocess.Popen(cmd, |
| stderr=subprocess.PIPE) |
| |
| # wait for devserver to start |
| current_time = time.time() |
| deadline = current_time + DEVSERVER_START_TIMEOUT |
| while current_time < deadline: |
| current_time = time.time() |
| try: |
| urllib2.urlopen(CHECK_HEALTH_URL, timeout=0.05) |
| break |
| except Exception: |
| continue |
| else: |
| self.fail('Devserver failed to start within timeout.') |
| |
| return process |
| |
| def VerifyHandleUpdate(self): |
| """Tests running the server and getting an update for the given protocol.""" |
| request = urllib2.Request(UPDATE_URL, UPDATE_REQUEST) |
| connection = urllib2.urlopen(request) |
| response = connection.read() |
| connection.close() |
| self.assertNotEqual('', response) |
| |
| # Parse the response and check if it contains the right result. |
| dom = minidom.parseString(response) |
| update = dom.getElementsByTagName('updatecheck')[0] |
| url = self.VerifyV3Response(update) |
| |
| # Try to fetch the image. |
| connection = urllib2.urlopen(url) |
| contents = connection.read() |
| connection.close() |
| self.assertEqual('Developers, developers, developers!\n', contents) |
| |
| def VerifyV3Response(self, update): |
| """Verifies the update DOM from a v3 response and returns the url.""" |
| # Parse the response and check if it contains the right result. |
| urls = update.getElementsByTagName('urls')[0] |
| url = urls.getElementsByTagName('url')[0] |
| |
| codebase = url.getAttribute('codebase') |
| self.assertEqual(SERVE_URL, codebase) |
| |
| manifest = update.getElementsByTagName('manifest')[0] |
| packages = manifest.getElementsByTagName('packages')[0] |
| package = packages.getElementsByTagName('package')[0] |
| filename = package.getAttribute('name') |
| self.assertEqual(TEST_IMAGE_NAME, filename) |
| |
| hash_value = package.getAttribute('hash') |
| self.assertEqual(EXPECTED_HASH, hash_value) |
| |
| url = os.path.join(codebase, filename) |
| return url |
| |
| # Tests begin here. |
| def testHandleUpdateV3(self): |
| self.VerifyHandleUpdate() |
| |
| def testApiBadSetNextUpdateRequest(self): |
| """Tests sending a bad setnextupdate request.""" |
| # Send bad request and ensure it fails... |
| try: |
| request = urllib2.Request(API_SET_UPDATE_URL, '') |
| connection = urllib2.urlopen(request) |
| connection.read() |
| connection.close() |
| self.fail('Invalid setnextupdate request did not fail!') |
| except urllib2.URLError: |
| pass |
| |
| def testApiBadSetNextUpdateURL(self): |
| """Tests contacting a bad setnextupdate url.""" |
| # Send bad request and ensure it fails... |
| try: |
| connection = urllib2.urlopen(API_SET_UPDATE_BAD_URL) |
| connection.read() |
| connection.close() |
| self.fail('Invalid setnextupdate url did not fail!') |
| except urllib2.URLError: |
| pass |
| |
| def testApiBadHostInfoURL(self): |
| """Tests contacting a bad hostinfo url.""" |
| # Send bad request and ensure it fails... |
| try: |
| connection = urllib2.urlopen(API_HOST_INFO_BAD_URL) |
| connection.read() |
| connection.close() |
| self.fail('Invalid hostinfo url did not fail!') |
| except urllib2.URLError: |
| pass |
| |
| def testApiHostInfoAndSetNextUpdate(self): |
| """Tests using the setnextupdate and hostinfo api commands.""" |
| # Send setnextupdate command. |
| request = urllib2.Request(API_SET_UPDATE_URL, API_SET_UPDATE_REQUEST) |
| connection = urllib2.urlopen(request) |
| response = connection.read() |
| connection.close() |
| |
| # Send hostinfo command and verify the setnextupdate worked. |
| connection = urllib2.urlopen(API_HOST_INFO_URL) |
| response = connection.read() |
| connection.close() |
| |
| self.assertEqual( |
| json.loads(response)['forced_update_label'], API_SET_UPDATE_REQUEST) |
| |
| def testPidFile(self): |
| """Test that using a pidfile works correctly.""" |
| with open(self.pidfile, 'r') as f: |
| pid = f.read() |
| |
| # Let's assert some process information about the devserver. |
| self.assertTrue(pid.isdigit()) |
| process = psutil.Process(int(pid)) |
| self.assertTrue(process.is_running()) |
| self.assertTrue('devserver.py' in process.cmdline) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |