recipe_modules/luci_config/test_api.py

# Copyright 2020 The Fuchsia 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 base64

from google.protobuf import text_format as textpb

from recipe_engine import recipe_test_api


class LuciConfigTestApi(recipe_test_api.RecipeTestApi):
    def mock_config(self, project, config_name, data, nesting=None):
        """Mock a config returned by the luci-config API.

        Args:
            project (str): The LUCI project name.
            config_name (str): The name of the config file to mock, e.g.
                "commit-queue.cfg".
            data (str or protobuf): The mock data that should be returned.
                Either a string containing a textproto, or a protobuf object.
            nesting (str): Parent step under which this step is nested.
        """
        if not isinstance(data, str):
            data = textpb.MessageToString(data)
        response = {"content": base64.b64encode(data)}
        step_name = "fetch %s %s.get" % (project, config_name)
        if nesting:  # pragma: no cover
            step_name = "%s.%s" % (nesting, step_name)
        return self.m.url.json(step_name, response)

    def mock_local_config(self, project, config_name, data, nesting=None):
        """Mock a config read from disk.

        Args:
            project (str): The LUCI project name.
            config_name (str): The name of the config file to mock, e.g.
                "commit-queue.cfg".
            data (str or protobuf): The mock data that should be returned.
                Either a string containing a textproto, or a protobuf object.
            nesting (str): Parent step under which this step is nested.
        """
        if not isinstance(data, str):
            data = textpb.MessageToString(data)
        step_name = "read %s %s" % (project, config_name)
        if nesting:  # pragma: no cover
            step_name = "%s.%s" % (nesting, step_name)
        return self.step_data(step_name, self.m.file.read_text(data))