blob: fdeecab9c71f0556a2b2e0c4335315ee01ba0023 [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright 2019 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.
"""Unit tests for the mob-remote-scheduler"""
import unittest
from unittest import mock
from moblab_common import devserver_connector
class DevserverConnectorTest(unittest.TestCase):
"""Tests for the devserver connector."""
def setUp(self):
self.dev_conn = devserver_connector.DevserverConnector(
("test_storage_bucket")
)
@mock.patch("moblab_common.devserver_connector.requests.get")
def test_devserver_command(self, mock_get_request):
response = mock.Mock()
response.status_code = 200
mock_get_request.return_value = response
result = self.dev_conn.send_devserver_command(
"testcommand", "test_params"
)
self.assertTrue(result)
mock_get_request.assert_called_once_with(
"http://devserver:8080/testcommand?test_params", timeout=600
)
@mock.patch("moblab_common.devserver_connector.requests.get")
def test_devserver_command_error(self, mock_get_request):
response = mock.Mock()
response.status_code = 302
mock_get_request.return_value = response
result = self.dev_conn.send_devserver_command(
"testcommand", "test_params"
)
self.assertFalse(result)
mock_get_request.assert_called_once_with(
"http://devserver:8080/testcommand?test_params", timeout=600
)
@mock.patch("moblab_common.devserver_connector.requests.get")
def test_devserver_command_raises(self, mock_get_request):
test_exception = (
devserver_connector.requests.exceptions.RequestException("Testing")
)
mock_get_request.side_effect = test_exception
result = self.dev_conn.send_devserver_command(
"testcommand", "test_params"
)
self.assertFalse(result)
mock_get_request.assert_called_once_with(
"http://devserver:8080/testcommand?test_params", timeout=600
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.send_devserver_command"
)
def test_stage_build_on_devserver(self, mock_send_devserver_command):
mock_send_devserver_command.return_value = True
result = self.dev_conn.stage_build_on_devserver(
"gs://testarchiveurl", "t1,t2"
)
mock_send_devserver_command.assert_called_once_with(
"stage", "archive_url=gs://testarchiveurl&artifacts=t1,t2"
)
self.assertTrue(result)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.send_devserver_command"
)
def test_check_build_staged_on_devserver(
self, mock_send_devserver_command
):
mock_send_devserver_command.return_value = True
result = self.dev_conn.check_build_staged_on_devserver(
"gs://testarchiveurl", "t1,t2"
)
mock_send_devserver_command.assert_called_once_with(
"is_staged", "archive_url=gs://testarchiveurl&artifacts=t1,t2"
)
self.assertTrue(result)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.check_build_staged_on_devserver"
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.stage_build_on_devserver"
)
@mock.patch("logging.info")
def test_stage_build(
self,
mock_logging,
mock_stage_build_on_devserver,
mock_check_build_staged_on_devserver,
):
mock_check_build_staged_on_devserver.side_effect = [False, True]
result = self.dev_conn.stage_build("testboard", "testbuild")
self.assertTrue(result)
mock_logging.assert_called_once()
mock_stage_build_on_devserver.assert_called_once_with(
"gs://test_storage_bucket/testboard-release/testbuild",
"full_payload,autotest_packages,stateful,quick_provision",
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.check_build_staged_on_devserver"
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.stage_build_on_devserver"
)
@mock.patch("logging.info")
def test_stage_build_custom_build(
self,
mock_logging,
mock_stage_build_on_devserver,
mock_check_build_staged_on_devserver,
):
mock_check_build_staged_on_devserver.side_effect = [False, True]
result = self.dev_conn.stage_build(
"testboard", "testbuild", "testcustombuild"
)
self.assertTrue(result)
mock_logging.assert_called_once()
mock_stage_build_on_devserver.assert_called_once_with(
"gs://test_storage_bucket/testboard-testcustombuild/testbuild",
"full_payload,autotest_packages,stateful,quick_provision",
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.check_build_staged_on_devserver"
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.stage_build_on_devserver"
)
@mock.patch("logging.info")
def test_stage_build_already_staged(
self,
mock_logging,
mock_stage_build_on_devserver,
mock_check_build_staged_on_devserver,
):
mock_check_build_staged_on_devserver.return_value = True
result = self.dev_conn.stage_build("testboard", "testbuild")
self.assertTrue(result)
mock_logging.assert_called_once()
self.assertEqual(0, mock_stage_build_on_devserver.call_count)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.check_build_staged_on_devserver"
)
@mock.patch(
"moblab_common.devserver_connector"
".DevserverConnector.stage_build_on_devserver"
)
@mock.patch("logging.info")
def test_stage_build_failed(
self,
mock_logging,
mock_stage_build_on_devserver,
mock_check_build_staged_on_devserver,
):
mock_check_build_staged_on_devserver.side_effect = [False, False]
result = self.dev_conn.stage_build("testboard", "testbuild")
self.assertFalse(result)
mock_logging.assert_called_once()
mock_stage_build_on_devserver.assert_called_once_with(
"gs://test_storage_bucket/testboard-release/testbuild",
"full_payload,autotest_packages,stateful,quick_provision",
)
if __name__ == "__main__":
unittest.main()