# 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. from __future__ import annotations import datetime as dt from unittest.mock import MagicMock, patch from crossbench.action_runner.action.enums import ReadyState from crossbench.action_runner.action.get import GetAction from crossbench.action_runner.action.meet_create import MeetCreateAction from crossbench.action_runner.action.meet_script import MeetScriptAction from crossbench.action_runner.default_action_runner import DefaultActionRunner from crossbench.action_runner.default_bond_action_runner import ( DefaultBondActionRunner) from crossbench.cli.config.secrets import ServiceAccount from tests import test_helper from tests.crossbench.base import BaseCrossbenchTestCase NOW_EPOCH = dt.datetime.now() class DefaultBondActionRunnerTestCase(BaseCrossbenchTestCase): SECRET = ServiceAccount.parse_dict({ "type": "fake", "project_id": "My Project", "private_key_id": "0badc0de", "private_key": "-----BEGIN PRIVATE KEY-----\n...", "client_email": "name@example.com", "client_id": "3", "auth_uri": "https://accounts.example.com/oauth2/auth", "token_uri": "https://oath2.example.com/token", "auth_provider_x509_cert_url": "https://example.com/oauth2/certs", "client_x509_cert_url": "https://example.com/x509", "universe_domain": "example.com", }) def _make_mock_run(self): mock_run = MagicMock(name="Mock Run") mock_run.secrets = MagicMock(name="Mock Secrets") mock_run.secrets.bond = self.SECRET return mock_run def _make_meet_create_mocks(self, mock_datetime, mock_bond_client_cls, create_meeting_duration=dt.timedelta(seconds=1), add_bots_duration=dt.timedelta(seconds=2), timeout=dt.timedelta(seconds=30)): mock_datetime.now.side_effect = [ NOW_EPOCH, # deadline init NOW_EPOCH + create_meeting_duration, # get timeout after create_meeting NOW_EPOCH + create_meeting_duration + add_bots_duration, # get timeout after add_bots ] mock_action_runner = MagicMock(name="Mock ActionRunner") mock_action_runner.get.side_effect = [None] bond_action_runner = DefaultBondActionRunner(mock_action_runner) mock_run = self._make_mock_run() mock_bond_client = mock_bond_client_cls.return_value mock_bond_client.create_meeting.side_effect = ["mock-conference-code"] mock_bond_client.add_bots.side_effect = [ [1, 2, 3], ] action = MeetCreateAction.parse_dict({ "action": "meet_create", "bots": { "num_of_bots": 3, "ttl_secs": 123, }, "timeout": timeout, }) return (mock_run, mock_bond_client, mock_action_runner, bond_action_runner, action) def test_get_current_conference_code(self): action_runner = DefaultActionRunner() bond_action_runner = DefaultBondActionRunner(action_runner) for browser in self.browsers: browser.set_current_url("https://meet.google.com/abc-def-ghi") code = bond_action_runner.get_current_conference_code(browser=browser) self.assertEqual(code, "abc-def-ghi") def test_get_current_conference_code_invalid(self): action_runner = DefaultActionRunner() bond_action_runner = DefaultBondActionRunner(action_runner) for browser in self.browsers: browser.set_current_url("https://www.google.com") with self.assertRaisesRegex(RuntimeError, "Unsupported URL for Bond action"): bond_action_runner.get_current_conference_code(browser=browser) @patch( "crossbench.action_runner.default_bond_action_runner.BondClient", autospec=True) @patch("crossbench.action_runner.default_bond_action_runner.dt.datetime") def test_meet_create(self, mock_datetime, mock_bond_client_cls): (mock_run, mock_bond_client, mock_action_runner, bond_action_runner, action) = self._make_meet_create_mocks(mock_datetime, mock_bond_client_cls) bond_action_runner.meet_create(mock_run, action) mock_bond_client.create_meeting.assert_called_once_with( timeout=dt.timedelta(seconds=30)) mock_bond_client.add_bots.assert_called_once_with( "mock-conference-code", action.bots, timeout=dt.timedelta(seconds=29)) mock_action_runner.get.assert_called_once_with( mock_run, GetAction( "https://meet.google.com/mock-conference-code", ready_state=ReadyState.COMPLETE, target=action.target, timeout=dt.timedelta(seconds=27))) @patch( "crossbench.action_runner.default_bond_action_runner.BondClient", autospec=True) @patch("crossbench.action_runner.default_bond_action_runner.dt.datetime") def test_meet_create_create_meeting_timeout(self, mock_datetime, mock_bond_client_cls): (mock_run, mock_bond_client, mock_action_runner, bond_action_runner, action) = self._make_meet_create_mocks( mock_datetime, mock_bond_client_cls, create_meeting_duration=dt.timedelta(seconds=30)) with self.assertRaises(TimeoutError): bond_action_runner.meet_create(mock_run, action) mock_bond_client.create_meeting.assert_called_once_with( timeout=dt.timedelta(seconds=30)) mock_bond_client.add_bots.assert_not_called() mock_action_runner.get.assert_not_called() @patch( "crossbench.action_runner.default_bond_action_runner.BondClient", autospec=True) @patch("crossbench.action_runner.default_bond_action_runner.dt.datetime") def test_meet_create_add_bots_timeout(self, mock_datetime, mock_bond_client_cls): (mock_run, mock_bond_client, mock_action_runner, bond_action_runner, action) = self._make_meet_create_mocks( mock_datetime, mock_bond_client_cls, create_meeting_duration=dt.timedelta(seconds=1), add_bots_duration=dt.timedelta(seconds=29)) with self.assertRaises(TimeoutError): bond_action_runner.meet_create(mock_run, action) mock_bond_client.create_meeting.assert_called_once_with( timeout=dt.timedelta(seconds=30)) mock_bond_client.add_bots.assert_called_once_with( "mock-conference-code", action.bots, timeout=dt.timedelta(seconds=29)) mock_action_runner.get.assert_not_called() @patch( "crossbench.action_runner.default_bond_action_runner.BondClient", autospec=True) def test_meet_script(self, mock_bond_client_cls): bond_action_runner = DefaultBondActionRunner( MagicMock(name="Mock ActionRunner")) mock_run = self._make_mock_run() mock_run.browser.current_url = "https://meet.google.com/abc-def-ghi" action = MeetScriptAction.parse_dict({ "action": "meet_script", "script": "test script", "timeout": 17 }) mock_bond_client = mock_bond_client_cls.return_value mock_bond_client.run_script.side_effect = [None] bond_action_runner.meet_script(mock_run, action) mock_bond_client.run_script.assert_called_once_with( "abc-def-ghi", "test script", dt.timedelta(seconds=17)) if __name__ == "__main__": test_helper.run_pytest(__file__)