# Lint as: python2, python3 # Copyright (c) 2013 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. from __future__ import absolute_import from __future__ import division from __future__ import print_function from six.moves import range import six import json import mox import time import unittest from six.moves import urllib import common from autotest_lib.client.common_lib import global_config from autotest_lib.server import site_utils _DEADBUILD = 'deadboard-release/R33-4966.0.0' _LIVEBUILD = 'liveboard-release/R32-4920.14.0' _STATUS_TEMPLATE = ''' { "username": "fizzbin@google.com", "date": "2013-11-16 00:25:23.511208", "message": "%s", "can_commit_freely": %s, "general_state": "%s" } ''' def _make_status(message, can_commit, state): return _STATUS_TEMPLATE % (message, can_commit, state) def _make_open_status(message, state): return _make_status(message, 'true', state) def _make_closed_status(message): return _make_status(message, 'false', 'closed') def _make_deadbuild_status(message): return _make_status(message, 'false', 'open') _OPEN_STATUS_VALUES = [ _make_open_status('Lab is up (cross your fingers)', 'open'), _make_open_status('Lab is on fire', 'throttled'), _make_open_status('Lab is up despite deadboard', 'open'), _make_open_status('Lab is up despite .*/R33-4966.0.0', 'open'), ] _CLOSED_STATUS_VALUES = [ _make_closed_status('Lab is down for spite'), _make_closed_status('Lab is down even for [%s]' % _LIVEBUILD), _make_closed_status('Lab is down even for [%s]' % _DEADBUILD), ] _DEADBUILD_STATUS_VALUES = [ _make_deadbuild_status('Lab is up except for [deadboard-]'), _make_deadbuild_status('Lab is up except for [board- deadboard-]'), _make_deadbuild_status('Lab is up except for [.*/R33-]'), _make_deadbuild_status('Lab is up except for [deadboard-.*/R33-]'), _make_deadbuild_status('Lab is up except for [ deadboard-]'), _make_deadbuild_status('Lab is up except for [deadboard- ]'), _make_deadbuild_status('Lab is up [first .*/R33- last]'), _make_deadbuild_status('liveboard is good, but [deadboard-] is bad'), _make_deadbuild_status('Lab is up [deadboard- otherboard-]'), _make_deadbuild_status('Lab is up [otherboard- deadboard-]'), ] _FAKE_URL = 'ignore://not.a.url' class _FakeURLResponse(object): """Everything needed to pretend to be a response from urlopen(). Creates a StringIO instance to handle the File operations. N.B. StringIO is lame: we can't inherit from it (super won't work), and it doesn't implement __getattr__(), either. So, we have to manually forward calls to the StringIO object. This forwards only what empirical testing says is required; YMMV. """ def __init__(self, code, buffer): self._stringio = six.StringIO(buffer) self._code = code def read(self, size=-1): """Standard file-like read operation. @param size size for read operation. """ return self._stringio.read(size) def getcode(self): """Get URL HTTP response code.""" return self._code class GetStatusTest(mox.MoxTestBase): """Test case for _get_lab_status(). We mock out dependencies on urllib2 and time.sleep(), and confirm that the function returns the proper JSON representation for a pre-defined response. """ def setUp(self): super(GetStatusTest, self).setUp() self.mox.StubOutWithMock(urllib.request, 'urlopen') self.mox.StubOutWithMock(time, 'sleep') def test_success(self): """Test that successful calls to urlopen() succeed.""" json_string = _OPEN_STATUS_VALUES[0] json_value = json.loads(json_string) urllib.request.urlopen(mox.IgnoreArg()).AndReturn( _FakeURLResponse(200, json_string)) self.mox.ReplayAll() result = site_utils._get_lab_status(_FAKE_URL) self.mox.VerifyAll() self.assertEqual(json_value, result) def test_retry_ioerror(self): """Test that an IOError retries at least once.""" json_string = _OPEN_STATUS_VALUES[0] json_value = json.loads(json_string) urllib.request.urlopen(mox.IgnoreArg()).AndRaise( IOError('Fake I/O error for a fake URL')) time.sleep(mox.IgnoreArg()).AndReturn(None) urllib.request.urlopen(mox.IgnoreArg()).AndReturn( _FakeURLResponse(200, json_string)) self.mox.ReplayAll() result = site_utils._get_lab_status(_FAKE_URL) self.mox.VerifyAll() self.assertEqual(json_value, result) def test_retry_http_internal_error(self): """Test that an HTTP error retries at least once.""" json_string = _OPEN_STATUS_VALUES[0] json_value = json.loads(json_string) urllib.request.urlopen(mox.IgnoreArg()).AndReturn( _FakeURLResponse(500, '')) time.sleep(mox.IgnoreArg()).AndReturn(None) urllib.request.urlopen(mox.IgnoreArg()).AndReturn( _FakeURLResponse(200, json_string)) self.mox.ReplayAll() result = site_utils._get_lab_status(_FAKE_URL) self.mox.VerifyAll() self.assertEqual(json_value, result) def test_failure_ioerror(self): """Test that there's a failure if urlopen() never succeeds.""" json_string = _OPEN_STATUS_VALUES[0] json_value = json.loads(json_string) for _ in range(site_utils._MAX_LAB_STATUS_ATTEMPTS): urllib.request.urlopen(mox.IgnoreArg()).AndRaise( IOError('Fake I/O error for a fake URL')) time.sleep(mox.IgnoreArg()).AndReturn(None) self.mox.ReplayAll() result = site_utils._get_lab_status(_FAKE_URL) self.mox.VerifyAll() self.assertEqual(None, result) def test_failure_http_internal_error(self): """Test that there's a failure for a permanent HTTP error.""" json_string = _OPEN_STATUS_VALUES[0] json_value = json.loads(json_string) for _ in range(site_utils._MAX_LAB_STATUS_ATTEMPTS): urllib.request.urlopen(mox.IgnoreArg()).AndReturn( _FakeURLResponse(404, 'Not here, never gonna be')) time.sleep(mox.IgnoreArg()).InAnyOrder().AndReturn(None) self.mox.ReplayAll() result = site_utils._get_lab_status(_FAKE_URL) self.mox.VerifyAll() self.assertEqual(None, result) class DecodeStatusTest(unittest.TestCase): """Test case for _decode_lab_status(). Testing covers three distinct possible states: 1. Lab is up. All calls to _decode_lab_status() will succeed without raising an exception. 2. Lab is down. All calls to _decode_lab_status() will fail with TestLabException. 3. Build disabled. Calls to _decode_lab_status() will succeed, except that board `_DEADBUILD` will raise TestLabException. """ def _assert_lab_open(self, lab_status): """Test that open status values are handled properly. Test that _decode_lab_status() succeeds when the lab status is up. @param lab_status JSON value describing lab status. """ site_utils._decode_lab_status(lab_status, _LIVEBUILD) site_utils._decode_lab_status(lab_status, _DEADBUILD) def _assert_lab_closed(self, lab_status): """Test that closed status values are handled properly. Test that _decode_lab_status() raises TestLabException when the lab status is down. @param lab_status JSON value describing lab status. """ with self.assertRaises(site_utils.TestLabException): site_utils._decode_lab_status(lab_status, _LIVEBUILD) with self.assertRaises(site_utils.TestLabException): site_utils._decode_lab_status(lab_status, _DEADBUILD) def _assert_lab_deadbuild(self, lab_status): """Test that disabled builds are handled properly. Test that _decode_lab_status() raises TestLabException for build `_DEADBUILD` and succeeds otherwise. @param lab_status JSON value describing lab status. """ site_utils._decode_lab_status(lab_status, _LIVEBUILD) with self.assertRaises(site_utils.TestLabException): site_utils._decode_lab_status(lab_status, _DEADBUILD) def _assert_lab_status(self, test_values, checker): """General purpose test for _decode_lab_status(). Decode each JSON string in `test_values`, and call the `checker` function to test the corresponding status is correctly handled. @param test_values Array of JSON encoded strings representing lab status. @param checker Function to be called against each of the lab status values in the `test_values` array. """ for s in test_values: lab_status = json.loads(s) checker(lab_status) def test_open_lab(self): """Test that open lab status values are handled correctly.""" self._assert_lab_status(_OPEN_STATUS_VALUES, self._assert_lab_open) def test_closed_lab(self): """Test that closed lab status values are handled correctly.""" self._assert_lab_status(_CLOSED_STATUS_VALUES, self._assert_lab_closed) def test_dead_build(self): """Test that disabled builds are handled correctly.""" self._assert_lab_status(_DEADBUILD_STATUS_VALUES, self._assert_lab_deadbuild) class CheckStatusTest(mox.MoxTestBase): """Test case for `check_lab_status()`. We mock out dependencies on `global_config.global_config()`, `_get_lab_status()` and confirm that the function succeeds or fails as expected. N.B. We don't mock `_decode_lab_status()`; if DecodeStatusTest is failing, this test may fail, too. """ def setUp(self): super(CheckStatusTest, self).setUp() self.mox.StubOutWithMock(global_config.global_config, 'get_config_value') self.mox.StubOutWithMock(site_utils, '_get_lab_status') def _setup_not_cautotest(self): """Set up to mock the "we're not on cautotest" case.""" global_config.global_config.get_config_value( 'SERVER', 'hostname').AndReturn('not-cautotest') def _setup_no_status(self): """Set up to mock lab status as unavailable.""" global_config.global_config.get_config_value( 'SERVER', 'hostname').AndReturn('cautotest') global_config.global_config.get_config_value( 'CROS', 'lab_status_url').AndReturn(_FAKE_URL) site_utils._get_lab_status(_FAKE_URL).AndReturn(None) def _setup_lab_status(self, json_string): """Set up to mock a given lab status. @param json_string JSON string for the JSON object to return from `_get_lab_status()`. """ global_config.global_config.get_config_value( 'SERVER', 'hostname').AndReturn('cautotest') global_config.global_config.get_config_value( 'CROS', 'lab_status_url').AndReturn(_FAKE_URL) json_value = json.loads(json_string) site_utils._get_lab_status(_FAKE_URL).AndReturn(json_value) def _try_check_status(self, build): """Test calling check_lab_status() with `build`.""" try: self.mox.ReplayAll() site_utils.check_lab_status(build) finally: self.mox.VerifyAll() def test_non_cautotest(self): """Test a call with a build when the host isn't cautotest.""" self._setup_not_cautotest() self._try_check_status(_LIVEBUILD) def test_no_lab_status(self): """Test with a build when `_get_lab_status()` returns `None`.""" self._setup_no_status() self._try_check_status(_LIVEBUILD) def test_lab_up_live_build(self): """Test lab open with a build specified.""" self._setup_lab_status(_OPEN_STATUS_VALUES[0]) self._try_check_status(_LIVEBUILD) def test_lab_down_live_build(self): """Test lab closed with a build specified.""" self._setup_lab_status(_CLOSED_STATUS_VALUES[0]) with self.assertRaises(site_utils.TestLabException): self._try_check_status(_LIVEBUILD) def test_build_disabled_live_build(self): """Test build disabled with a live build specified.""" self._setup_lab_status(_DEADBUILD_STATUS_VALUES[0]) self._try_check_status(_LIVEBUILD) def test_build_disabled_dead_build(self): """Test build disabled with the disabled build specified.""" self._setup_lab_status(_DEADBUILD_STATUS_VALUES[0]) with self.assertRaises(site_utils.TestLabException): self._try_check_status(_DEADBUILD) if __name__ == '__main__': unittest.main()