utils/side_effects/config_loader_unittest.py

# 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.

import json
import os
import shutil
import unittest
import tempfile

import common
from autotest_lib.utils.side_effects import config_loader
from autotest_lib.utils.side_effects.proto import config_pb2


class LoadTestCase(unittest.TestCase):
    """Test loading side_effects_config.json."""

    def setUp(self):
        self._results_dir = tempfile.mkdtemp()

    def tearDown(self):
        shutil.rmtree(self._results_dir)

    def test_missing_config_file(self):
        parsed_config = config_loader.load(self._results_dir)
        self.assertIsNone(parsed_config)

    def test_full_config(self):
        config = {
            'tko': {
                'proxy_socket': '/file-system/foo-socket',
                'mysql_user': 'foo-user',
                'mysql_password_file': '/file-system/foo-password-file'
            },
            'google_storage': {
                'bucket': 'foo-bucket',
                'credentials_file': '/file-system/foo-creds'
            },
            'this_field_is_ignored': True
        }
        path = os.path.join(self._results_dir,
                            'side_effects_config.json')
        with open(path, 'w') as f:
            f.write(json.dumps(config))
        parsed_config = config_loader.load(self._results_dir)

        self.assertEqual(parsed_config.tko.proxy_socket,
                         '/file-system/foo-socket')
        self.assertEqual(parsed_config.tko.mysql_user, 'foo-user')
        self.assertEqual(parsed_config.tko.mysql_password_file,
                         '/file-system/foo-password-file')
        self.assertEqual(parsed_config.google_storage.bucket, 'foo-bucket')
        self.assertEqual(parsed_config.google_storage.credentials_file,
                         '/file-system/foo-creds')
        self.assertEqual(parsed_config.chrome_perf.enabled, False)
        self.assertEqual(parsed_config.cts.enabled, False)

    def test_bad_config(self):
        path = os.path.join(self._results_dir,
                            'side_effects_config.json')
        with open(path, 'w') as f:
            f.write('Not a JSON, hence unparseable')

        with self.assertRaisesRegexp(Exception, 'JSON'):
            parsed_config = config_loader.load(self._results_dir)


class ValidateTKOTestCase(unittest.TestCase):
    """Test validating the TKO-related fields of a side_effects.Config proto."""

    def setUp(self):
        self._tempdir = tempfile.mkdtemp()
        self._config = config_pb2.Config(
            tko=config_pb2.TKOConfig(
                proxy_socket=_tempfile(self._tempdir),
                mysql_user='foo-user',
                mysql_password_file=_tempfile(self._tempdir)
            ))

    def tearDown(self):
        shutil.rmtree(self._tempdir)

    def test_complete_config(self):
        config_loader.validate_tko(self._config)

    def test_missing_proxy_socket_path(self):
        self._config.tko.proxy_socket = ''
        with self.assertRaisesRegexp(ValueError, 'proxy socket'):
            config_loader.validate_tko(self._config)

    def test_missing_mysql_user(self):
        self._config.tko.mysql_user = ''
        with self.assertRaisesRegexp(ValueError, 'MySQL user'):
            config_loader.validate_tko(self._config)

    def test_missing_mysql_password_file_path(self):
        self._config.tko.mysql_password_file = ''
        with self.assertRaisesRegexp(ValueError, 'MySQL password'):
            config_loader.validate_tko(self._config)

    def test_missing_proxy_socket_file(self):
        os.remove(self._config.tko.proxy_socket)
        with self.assertRaisesRegexp(OSError, 'proxy socket'):
            config_loader.validate_tko(self._config)

    def test_missing_mysql_password_file(self):
        os.remove(self._config.tko.mysql_password_file)
        with self.assertRaisesRegexp(OSError, 'MySQL password'):
            config_loader.validate_tko(self._config)


class ValidateGoogleStorageTestCase(unittest.TestCase):
    """Test validating the TKO-related fields of a side_effects.Config proto."""

    def setUp(self):
        self._tempdir = tempfile.mkdtemp()
        self._config = config_pb2.Config(
            google_storage=config_pb2.GoogleStorageConfig(
                bucket='foo-bucket',
                credentials_file=_tempfile(self._tempdir)
            ))

    def tearDown(self):
        shutil.rmtree(self._tempdir)

    def test_complete_config(self):
        config_loader.validate_google_storage(self._config)

    def test_missing_bucket(self):
        self._config.google_storage.bucket = ''
        with self.assertRaisesRegexp(ValueError, 'bucket'):
            config_loader.validate_google_storage(self._config)

    def test_missing_credentials_file_path(self):
        self._config.google_storage.credentials_file = ''
        with self.assertRaisesRegexp(ValueError, 'credentials'):
            config_loader.validate_google_storage(self._config)

    def test_missing_credentials_file(self):
        os.remove(self._config.google_storage.credentials_file)
        with self.assertRaisesRegexp(OSError, 'credentials'):
            config_loader.validate_google_storage(self._config)


def _tempfile(tempdir):
    _, name = tempfile.mkstemp(dir=tempdir)
    return name


if __name__ == '__main__':
    unittest.main()