#!/usr/bin/env vpython3
# Copyright 2025 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unittest for google_storage_helper.py.

Example usage:
  vpython3 google_storage_helper_test.py
"""

import os
import sys
import time
import unittest
from unittest import mock

import google_storage_helper as helper  # pylint: disable=import-error
from parameterized import parameterized  # pylint: disable=import-error

LIB_PATH = os.path.join(os.path.dirname(__file__), os.pardir, os.pardir)
sys.path.append(LIB_PATH)

DIR_SOURCE_ROOT = os.environ.get(
    'CHECKOUT_SOURCE_ROOT',
    os.path.abspath(
        os.path.join(
            os.path.dirname(__file__),
            os.pardir,
            os.pardir,
            os.pardir,
            os.pardir,
        )
    ),
)
DEVIL_PATH = os.path.join(DIR_SOURCE_ROOT, 'third_party', 'catapult', 'devil')

if DEVIL_PATH not in sys.path:
    sys.path.append(DEVIL_PATH)
from devil.utils import cmd_helper


class GoogleStorageHelperTest(unittest.TestCase):
    @parameterized.expand(
        [
            (
                'empty_bucket_name',
                '',
                '',
            ),
            (
                'bucket_name_with_gs',
                'gs://foo/bar',
                'foo/bar',
            ),
            (
                'bucket_name_with_gs_end_slash',
                'gs://foo/bar/',
                'foo/bar',
            ),
            (
                'bucket_name_no_gs',
                'foo/bar',
                'foo/bar',
            ),
            (
                'bucket_name_no_gs_end_slash',
                'gs://foo/bar/',
                'foo/bar',
            ),
        ]
    )
    def test_format_bucket_name(self, _, bucket, expected):
        got = helper._format_bucket_name(bucket)  # pylint: disable=protected-access
        self.assertEqual(
            got,
            expected,
        )

    @mock.patch('platform.system', autospec=True)
    @mock.patch.object(cmd_helper, 'RunCmd', autospec=True)
    def test_exists(self, mock_cmd_helper, mock_system):
        bucket = 'foo'
        name = 'bar'
        with self.subTest(name='Windows'):
            with mock.patch.object(
                helper,
                'get_gsutil_script_path',
                autospec=True,
                return_value='path_to_gsutil_py',
            ):
                mock_system.return_value = 'Windows'
                helper.exists(name, bucket)
                mock_cmd_helper.assert_called_once_with(
                    ['path_to_gsutil_py', '-q', 'stat', 'gs://foo/bar']
                )
        mock_cmd_helper.reset_mock()
        with self.subTest(name='Linux'):
            with mock.patch.object(
                helper,
                'get_gsutil_script_path',
                autospec=True,
                return_value='path_to_gsutil',
            ):
                mock_system.return_value = 'Linux'
                helper.exists(name, bucket)
                mock_cmd_helper.assert_called_once_with(
                    ['path_to_gsutil', '-q', 'stat', 'gs://foo/bar']
                )

    @mock.patch('platform.system', autospec=True)
    @mock.patch.object(cmd_helper, 'RunCmd', autospec=True)
    @mock.patch.object(helper, 'get_url_link', autospec=True)
    def test_upload(self, mock_get_url_link, mock_cmd_helper, mock_system):
        bucket = 'foo'
        name = 'bar'
        filepath = os.path.join('abc', 'def.json')
        with self.subTest(name='Windows'):
            with mock.patch.object(
                helper,
                'get_gsutil_script_path',
                autospec=True,
                return_value='path_to_gsutil_py',
            ):
                mock_system.return_value = 'Windows'
                helper.upload(name, filepath, bucket)
                mock_cmd_helper.assert_called_once_with(
                    [
                        'path_to_gsutil_py',
                        '-q',
                        'cp',
                        os.path.join('abc', 'def.json'),
                        'gs://foo/bar',
                    ]
                )
                mock_get_url_link.assert_called_once_with(name, bucket, True)
        mock_cmd_helper.reset_mock()
        mock_get_url_link.reset_mock()
        with self.subTest(name='Linux'):
            with mock.patch.object(
                helper,
                'get_gsutil_script_path',
                autospec=True,
                return_value='path_to_gsutil',
            ):
                mock_system.return_value = 'Linux'
                helper.upload(name, filepath, bucket)
                mock_cmd_helper.assert_called_once_with(
                    [
                        'path_to_gsutil',
                        '-q',
                        'cp',
                        os.path.join('abc', 'def.json'),
                        'gs://foo/bar',
                    ]
                )
                mock_get_url_link.assert_called_once_with(name, bucket, True)

    @parameterized.expand(
        [
            (
                'empty_bucket_name',
                '',
                'bar',
                'https://storage.cloud.google.com//bar',
            ),
            ('empty_name', 'foo', '', 'https://storage.cloud.google.com/foo/'),
            (
                'normal',
                'foo',
                'bar',
                'https://storage.cloud.google.com/foo/bar',
            ),
        ]
    )
    def test_get_url_link(self, _, bucket, name, expected):
        got = helper.get_url_link(name, bucket, True)
        self.assertEqual(expected, got)

    @mock.patch('time.gmtime')
    def test_unique_name(self, mock_gmtime):
        basename = 'foo'
        suffix = '.json'
        expected = 'foo_2026_01_15_T12_30_00-UTC.json'
        mock_gmtime.return_value = time.struct_time(
            (2026, 1, 15, 12, 30, 0, 3, 15, 0)
        )
        got = helper.unique_name(basename, suffix)
        self.assertEqual(expected, got)

    @parameterized.expand(
        [
            ('empty_link', '', ['path_to_gsutil', '-q', 'cat', 'gs://']),
            (
                'valid gs_link',
                '/foo',
                ['path_to_gsutil', '-q', 'cat', 'gs://foo'],
            ),
            (
                'no_initial_slash',
                'foo/bar',
                ['path_to_gsutil', '-q', 'cat', 'gs://foo/bar'],
            ),
        ]
    )  # pylint: disable=no-self-use
    def test_read_from_link(self, _, link, expected_sequence):
        with mock.patch('platform.system', autospec=True, return_value='Linux'):
            with mock.patch.object(
                helper,
                'get_gsutil_script_path',
                autospec=True,
                return_value='path_to_gsutil',
            ):
                with mock.patch.object(
                    cmd_helper, 'GetCmdOutput', autospec=True
                ) as mock_get_cmd_output:
                    helper.read_from_link(link)
                    mock_get_cmd_output.assert_called_once_with(
                        expected_sequence
                    )


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