build/rbe/bbtool_test.py

#!/usr/bin/env fuchsia-vendored-python
# Copyright 2023 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 contextlib
import io
import unittest
from pathlib import Path
from unittest import mock

import bbtool
import cl_utils


class MainArgParserTests(unittest.TestCase):
    def test_fetch_reproxy_log(self):
        args = bbtool._MAIN_ARG_PARSER.parse_args(
            ["--bbid", "1234", "fetch_reproxy_log"]
        )
        self.assertEqual(args.bbid, "1234")
        self.assertFalse(args.verbose)


class BuildBucketToolTests(unittest.TestCase):
    def test_get_json_fields_success(self):
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(
            cl_utils,
            "subprocess_call",
            return_value=cl_utils.SubprocessResult(0, stdout=["{}"]),
        ) as mock_call:
            bb_json = bb.get_json_fields("5432")
            self.assertEqual(bb_json, dict())
        mock_call.assert_called_once()

    def test_get_json_fields_error(self):
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(
            cl_utils,
            "subprocess_call",
            return_value=cl_utils.SubprocessResult(1, stderr=["oops"]),
        ) as mock_call:
            with self.assertRaises(bbtool.BBError):
                bb.get_json_fields("6432")
        mock_call.assert_called_once()

    def test_download_reproxy_log_success(self):
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(
            cl_utils,
            "subprocess_call",
            return_value=cl_utils.SubprocessResult(0),
        ) as mock_call:
            self.assertEqual(
                bb.download_reproxy_log("1271", "reproxy_3.1415926.rrpl"), "\n"
            )
        mock_call.assert_called_once()

    def test_download_reproxy_log_error(self):
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(
            cl_utils,
            "subprocess_call",
            return_value=cl_utils.SubprocessResult(1),
        ) as mock_call:
            with self.assertRaises(bbtool.BBError):
                bb.download_reproxy_log("1271", "reproxy_3.1415926.rrpl")
        mock_call.assert_called_once()

    def test_fetch_reproxy_log_cached_already_cached(self):
        log_name = "reproxy_1.4142.rrpl"
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(Path, "mkdir") as mock_mkdir:
            with mock.patch.object(
                Path, "exists", return_value=True
            ) as mock_exists:
                cached_path = bb.fetch_reproxy_log_cached("132461", log_name)
                self.assertEqual(cached_path.name, log_name)  # in some temp dir
        mock_mkdir.assert_called_once()
        mock_exists.assert_called_with()

    def test_fetch_reproxy_log_cached_download_success(self):
        log_name = "reproxy_9.6142.rrpl"
        log_contents = "\n"
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(Path, "mkdir") as mock_mkdir:
            with mock.patch.object(
                Path, "exists", return_value=False
            ) as mock_exists:
                with mock.patch.object(
                    bbtool.BuildBucketTool,
                    "download_reproxy_log",
                    return_value=log_contents,
                ) as mock_download:
                    with mock.patch.object(Path, "write_text") as mock_write:
                        cached_path = bb.fetch_reproxy_log_cached(
                            "662481", log_name
                        )
                        self.assertEqual(
                            cached_path.name, log_name
                        )  # in some temp dir
        mock_mkdir.assert_called_once()
        mock_exists.assert_called_with()
        mock_write.assert_called_with(log_contents)

    def test_fetch_reproxy_log_cached_download_error(self):
        log_name = "reproxy_7.0123.rrpl"
        bb = bbtool.BuildBucketTool()
        with mock.patch.object(Path, "mkdir") as mock_mkdir:
            with mock.patch.object(
                Path, "exists", return_value=False
            ) as mock_exists:
                with mock.patch.object(
                    bbtool.BuildBucketTool,
                    "download_reproxy_log",
                    side_effect=bbtool.BBError("download error"),
                ) as mock_download:
                    with self.assertRaises(bbtool.BBError):
                        bb.fetch_reproxy_log_cached("662481", log_name)
        mock_mkdir.assert_called_once()
        mock_exists.assert_called_with()

    def test_get_rbe_build_info_no_child(self):
        bb = bbtool.BuildBucketTool()
        bbid = "8888"
        build_json = {"output": {"properties": {}}}
        with mock.patch.object(
            bbtool.BuildBucketTool, "get_json_fields", return_value=build_json
        ) as mock_get_json:
            rbe_build_id, rbe_build_json = bb.get_rbe_build_info(bbid)
            self.assertEqual(rbe_build_id, bbid)
            self.assertEqual(rbe_build_json, build_json)

    def test_get_rbe_build_info_with_child(self):
        bb = bbtool.BuildBucketTool()
        parent_bbid = "9977"
        child_bbid = "0404"
        parent_build_json = {
            "output": {"properties": {"child_build_id": child_bbid}}
        }
        child_build_json = {
            "output": {"properties": {"rpl_files": ["foo.rrpl"]}}
        }
        with mock.patch.object(
            bbtool.BuildBucketTool,
            "get_json_fields",
            side_effect=[parent_build_json, child_build_json],
        ) as mock_get_json:
            rbe_build_id, rbe_build_json = bb.get_rbe_build_info(parent_bbid)
            self.assertEqual(rbe_build_id, child_bbid)
            self.assertEqual(rbe_build_json, child_build_json)


class FetchReproxyLogFromBbidTests(unittest.TestCase):
    def test_lookup_and_fetch(self):
        bb_json = {
            "output": {"properties": {"rpl_files": ["reproxy_2.718.rrpl"]}}
        }
        reproxy_log_path = Path("/some/where/in/temp/foo.rrpl")
        with mock.patch.object(
            bbtool.BuildBucketTool,
            "get_rbe_build_info",
            return_value=("8728721", bb_json),
        ) as mock_rbe_build_info:
            with mock.patch.object(
                bbtool.BuildBucketTool,
                "fetch_reproxy_log_cached",
                return_value=reproxy_log_path,
            ) as mock_fetch_log:
                self.assertEqual(
                    bbtool.fetch_reproxy_log_from_bbid(
                        bbpath=Path("bb"), bbid="b789789"
                    ),
                    reproxy_log_path,
                )
        mock_rbe_build_info.assert_called_once()
        mock_fetch_log.assert_called_once()

    def test_no_rpl_files(self):
        bb_json = {"output": {"properties": {}}}
        with mock.patch.object(
            bbtool.BuildBucketTool,
            "get_rbe_build_info",
            return_value=("8728721", bb_json),
        ) as mock_rbe_build_info:
            self.assertIsNone(
                bbtool.fetch_reproxy_log_from_bbid(
                    bbpath=Path("bb"), bbid="b789789"
                )
            )
        mock_rbe_build_info.assert_called_once()


class MainTests(unittest.TestCase):
    def test_e2e_bbid_to_log_only(self):
        bb = bbtool._BB_TOOL
        bbid = "b991918261"
        reproxy_log_path = Path("/path/to/cache/foobar.rrpl")
        result = io.StringIO()
        with contextlib.redirect_stdout(result):
            with mock.patch.object(
                bbtool,
                "fetch_reproxy_log_from_bbid",
                return_value=reproxy_log_path,
            ) as mock_fetch_log:
                self.assertEqual(
                    bbtool.main(["--bbid", bbid, "fetch_reproxy_log"]), 0
                )
        self.assertTrue(
            result.getvalue().endswith(
                f"reproxy log name: {reproxy_log_path}\n"
            )
        )
        mock_fetch_log.assert_called_once_with(
            bbpath=bb, bbid=bbid.lstrip("b"), verbose=False
        )

    def test_e2e_bb_error(self):
        bbid = "b99669966"
        with mock.patch.object(
            bbtool.BuildBucketTool,
            "get_rbe_build_info",
            side_effect=bbtool.BBError("error"),
        ) as mock_rbe_build_info:
            self.assertEqual(
                bbtool.main(["--bbid", bbid, "fetch_reproxy_log"]), 1
            )
        mock_rbe_build_info.assert_called_once_with(
            bbid.lstrip("b"), verbose=False
        )


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