Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use test_rpc_error method in locust

Best Python code snippet using locust

resultstore_client_test.py

Source:resultstore_client_test.py

...48    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)49    @patch(50        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'51    )52    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):53        stub = ResultStoreUploadStubMock()54        stub.CreateInvocation = MagicMock(return_value=DEFAULT_RPC_ERROR)55        client = initialize_client(FLAGS)56        res = client.create_invocation()57        self.assertEqual(res, DEFAULT_RPC_ERROR)58class CreateConfigurationTest(absltest.TestCase):59    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)60    @patch(61        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'62    )63    def test_create_configuration_success(self, ResultStoreUploadStubMock,64                                          gen_new_uuid_mock):65        stub = ResultStoreUploadStubMock()66        stub.CreateConfiguration = MagicMock(67            side_effect=create_return_configuration)68        client = initialize_client(FLAGS)69        config = get_default_configuration(TEST_INVOCATION_ID, TEST_CONFIG_ID)70        res = client.create_configuration(config)71        self.assertEqual(res.id.configuration_id, TEST_CONFIG_ID)72        self.assertEqual(73            res.name,74            "invocations/{}/configs/{}".format(TEST_INVOCATION_ID,75                                               TEST_CONFIG_ID))76    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)77    @patch(78        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'79    )80    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):81        stub = ResultStoreUploadStubMock()82        stub.CreateConfiguration = MagicMock(return_value=DEFAULT_RPC_ERROR)83        client = initialize_client(FLAGS)84        config = get_default_configuration(TEST_INVOCATION_ID, TEST_CONFIG_ID)85        res = client.create_configuration(config)86        self.assertEqual(res, DEFAULT_RPC_ERROR)87class CreateTargetTest(absltest.TestCase):88    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)89    @patch(90        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'91    )92    def test_create_target_success(self, ResultStoreUploadStubMock,93                                   gen_new_uuid_mock):94        stub = ResultStoreUploadStubMock()95        stub.CreateTarget = MagicMock(side_effect=create_return_target)96        client = initialize_client(FLAGS)97        target = get_default_target(TEST_INVOCATION_ID, TEST_TARGET_ID)98        res = client.create_target(target)99        self.assertEqual(res.id.target_id, TEST_TARGET_ID)100        self.assertEqual(101            res.name,102            "invocations/{}/targets/{}".format(TEST_INVOCATION_ID,103                                               TEST_TARGET_ID))104    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)105    @patch(106        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'107    )108    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):109        stub = ResultStoreUploadStubMock()110        stub.CreateTarget = MagicMock(return_value=DEFAULT_RPC_ERROR)111        client = initialize_client(FLAGS)112        target = get_default_target(TEST_INVOCATION_ID, TEST_TARGET_ID)113        res = client.create_target(target)114        self.assertEqual(res, DEFAULT_RPC_ERROR)115class CreateConfiguredTargetTest(absltest.TestCase):116    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)117    @patch(118        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'119    )120    def test_create_configured_target_success(self, ResultStoreUploadStubMock,121                                              gen_new_uuid_mock):122        stub = ResultStoreUploadStubMock()123        stub.CreateConfiguredTarget = MagicMock(124            side_effect=create_return_configured_target)125        client = initialize_client(FLAGS)126        configured_target = get_default_configured_target(127            TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID)128        res = client.create_configured_target(configured_target)129        self.assertEqual(res.id.target_id, TEST_TARGET_ID)130        self.assertEqual(res.id.configuration_id, TEST_CONFIG_ID)131        self.assertEqual(res.id.invocation_id, TEST_INVOCATION_ID)132        self.assertEqual(133            res.name, "invocations/{}/targets/{}/configuredTargets/{}".format(134                TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID))135    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)136    @patch(137        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'138    )139    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):140        stub = ResultStoreUploadStubMock()141        stub.CreateConfiguredTarget = MagicMock(return_value=DEFAULT_RPC_ERROR)142        client = initialize_client(FLAGS)143        configured_target = get_default_configured_target(144            TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID)145        res = client.create_configured_target(configured_target)146        self.assertEqual(res, DEFAULT_RPC_ERROR)147class CreateActionTest(absltest.TestCase):148    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)149    @patch(150        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'151    )152    def test_create_action_success(self, ResultStoreUploadStubMock,153                                   gen_new_uuid_mock):154        stub = ResultStoreUploadStubMock()155        stub.CreateAction = MagicMock(side_effect=create_return_action)156        client = initialize_client(FLAGS)157        action = get_default_action(TEST_INVOCATION_ID, TEST_TARGET_ID,158                                    TEST_CONFIG_ID, TEST_ACTION_ID)159        res = client.create_action(action)160        self.assertEqual(res.id.action_id, TEST_ACTION_ID)161        self.assertEqual(162            res.name,163            "invocations/{}/targets/{}/configuredTargets/{}/actions/{}".format(164                TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID,165                TEST_ACTION_ID))166    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)167    @patch(168        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'169    )170    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):171        stub = ResultStoreUploadStubMock()172        stub.CreateAction = MagicMock(return_value=DEFAULT_RPC_ERROR)173        client = initialize_client(FLAGS)174        action = get_default_action(TEST_INVOCATION_ID, TEST_TARGET_ID,175                                    TEST_CONFIG_ID, TEST_ACTION_ID)176        res = client.create_action(action)177        self.assertEqual(res, DEFAULT_RPC_ERROR)178class UpdateActionTest(absltest.TestCase):179    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)180    @patch(181        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'182    )183    def test_update_action_success(self, ResultStoreUploadStubMock,184                                   gen_new_uuid_mock):185        stub = ResultStoreUploadStubMock()186        stub.UpdateAction = MagicMock(side_effect=create_return_action)187        client = initialize_client(FLAGS)188        action = get_default_action(TEST_INVOCATION_ID, TEST_TARGET_ID,189                                    TEST_CONFIG_ID, TEST_ACTION_ID)190        res = client.update_action(action,191                                   ['timing.duration', 'status_attributes'])192        self.assertEqual(res.id.action_id, TEST_ACTION_ID)193        self.assertEqual(194            res.name,195            "invocations/{}/targets/{}/configuredTargets/{}/actions/{}".format(196                TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID,197                TEST_ACTION_ID))198    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)199    @patch(200        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'201    )202    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):203        stub = ResultStoreUploadStubMock()204        stub.UpdateAction = MagicMock(return_value=DEFAULT_RPC_ERROR)205        client = initialize_client(FLAGS)206        action = get_default_action(TEST_INVOCATION_ID, TEST_TARGET_ID,207                                    TEST_CONFIG_ID, TEST_ACTION_ID)208        res = client.update_action(action, ['test'])209        self.assertEqual(res, DEFAULT_RPC_ERROR)210    def test_missing_update_fields(self):211        client = initialize_client(FLAGS)212        try:213            client.update_action(None, None)214            self.fail('update_action passed with empty update fields')215        except Error as error:216            self.assertEqual(error.args[0], UPDATE_FIELD_ERROR_MESSAGE)217class UpdateConfiguredTargetTest(absltest.TestCase):218    def test_missing_update_fields(self):219        client = initialize_client(FLAGS)220        try:221            client.update_configured_target(None, None)222            self.fail(223                'update_configured_target passed with empty update fields')224        except Error as error:225            self.assertEqual(error.args[0], UPDATE_FIELD_ERROR_MESSAGE)226    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)227    @patch(228        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'229    )230    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):231        stub = ResultStoreUploadStubMock()232        stub.UpdateConfiguredTarget = MagicMock(return_value=DEFAULT_RPC_ERROR)233        client = initialize_client(FLAGS)234        configured_target = get_default_configured_target(235            TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID)236        res = client.update_configured_target(configured_target, ['test'])237        self.assertEqual(res, DEFAULT_RPC_ERROR)238    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)239    @patch(240        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'241    )242    def test_update_configured_target_success(self, ResultStoreUploadStubMock,243                                              gen_new_uuid_mock):244        stub = ResultStoreUploadStubMock()245        stub.UpdateConfiguredTarget = MagicMock(246            side_effect=create_return_configured_target)247        client = initialize_client(FLAGS)248        configured_target = get_default_configured_target(249            TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID)250        res = client.update_configured_target(251            configured_target, ['timing.duration', 'status_attributes'])252        self.assertEqual(res.id.target_id, TEST_TARGET_ID)253        self.assertEqual(res.id.configuration_id, TEST_CONFIG_ID)254        self.assertEqual(255            res.name, "invocations/{}/targets/{}/configuredTargets/{}".format(256                TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID))257class UpdateTargetTest(absltest.TestCase):258    def test_missing_update_fields(self):259        client = initialize_client(FLAGS)260        try:261            client.update_target(None, None)262            self.fail(263                'update_configured_target passed with empty update fields')264        except Error as error:265            self.assertEqual(error.args[0], UPDATE_FIELD_ERROR_MESSAGE)266    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)267    @patch(268        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'269    )270    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):271        stub = ResultStoreUploadStubMock()272        stub.UpdateTarget = MagicMock(return_value=DEFAULT_RPC_ERROR)273        client = initialize_client(FLAGS)274        target = get_default_target(TEST_INVOCATION_ID, TEST_TARGET_ID)275        res = client.update_target(target, ['test'])276        self.assertEqual(res, DEFAULT_RPC_ERROR)277    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)278    @patch(279        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'280    )281    def test_update_target_success(self, ResultStoreUploadStubMock,282                                   gen_new_uuid_mock):283        stub = ResultStoreUploadStubMock()284        stub.UpdateTarget = MagicMock(side_effect=create_return_target)285        client = initialize_client(FLAGS)286        target = get_default_target(TEST_INVOCATION_ID, TEST_TARGET_ID)287        res = client.update_target(target,288                                   ['timing.duration', 'status_attributes'])289        self.assertEqual(res.id.target_id, TEST_TARGET_ID)290        self.assertEqual(291            res.name,292            "invocations/{}/targets/{}".format(TEST_INVOCATION_ID,293                                               TEST_TARGET_ID))294class FinalizeConfiguredTargetTest(absltest.TestCase):295    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)296    @patch(297        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'298    )299    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):300        stub = ResultStoreUploadStubMock()301        stub.FinalizeConfiguredTarget = MagicMock(302            return_value=DEFAULT_RPC_ERROR)303        client = initialize_client(FLAGS)304        res = client.finalize_configured_target(TEST_INVOCATION_ID,305                                                TEST_TARGET_ID, TEST_CONFIG_ID)306        self.assertEqual(res, DEFAULT_RPC_ERROR)307    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)308    @patch(309        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'310    )311    def test_finalize_configured_target_success(self,312                                                ResultStoreUploadStubMock,313                                                gen_new_uuid_mock):314        stub = ResultStoreUploadStubMock()315        stub.FinalizeConfiguredTarget = MagicMock(316            side_effect=create_finalized_configured_target)317        client = initialize_client(FLAGS)318        res = client.finalize_configured_target(TEST_INVOCATION_ID,319                                                TEST_TARGET_ID, TEST_CONFIG_ID)320        self.assertEqual(res.id.target_id, TEST_TARGET_ID)321        self.assertEqual(res.id.configuration_id, TEST_CONFIG_ID)322        self.assertEqual(res.id.invocation_id, TEST_INVOCATION_ID)323        self.assertEqual(324            res.name, "invocations/{}/targets/{}/configuredTargets/{}".format(325                TEST_INVOCATION_ID, TEST_TARGET_ID, TEST_CONFIG_ID))326class FinalizeTargetTest(absltest.TestCase):327    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)328    @patch(329        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'330    )331    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):332        stub = ResultStoreUploadStubMock()333        stub.FinalizeTarget = MagicMock(return_value=DEFAULT_RPC_ERROR)334        client = initialize_client(FLAGS)335        res = client.finalize_target(TEST_INVOCATION_ID, TEST_TARGET_ID)336        self.assertEqual(res, DEFAULT_RPC_ERROR)337    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)338    @patch(339        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'340    )341    def test_finalize_target_success(self, ResultStoreUploadStubMock,342                                     gen_new_uuid_mock):343        stub = ResultStoreUploadStubMock()344        stub.FinalizeTarget = MagicMock(side_effect=create_finalized_target)345        client = initialize_client(FLAGS)346        res = client.finalize_target(TEST_INVOCATION_ID, TEST_TARGET_ID)347        self.assertEqual(res.id.target_id, TEST_TARGET_ID)348        self.assertEqual(res.id.invocation_id, TEST_INVOCATION_ID)349        self.assertEqual(350            res.name,351            "invocations/{}/targets/{}".format(TEST_INVOCATION_ID,352                                               TEST_TARGET_ID))353class SingleUploadTest(absltest.TestCase):354    @flagsaver.flagsaver355    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)356    @patch(357        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'358    )359    def test_finalize_target_success(self, ResultStoreUploadStubMock,360                                     gen_new_uuid_mock):361        FLAGS.invocation_id = TEST_INVOCATION_ID362        stub = ResultStoreUploadStubMock()363        with patch.object(BigStoreClient, 'upload_files_to_bigstore',364                          lambda self, files: []):365            stub.CreateConfiguration = MagicMock(366                side_effect=create_return_configuration)367            stub.CreateTarget = MagicMock(side_effect=create_return_target)368            stub.CreateConfiguredTarget = MagicMock(369                side_effect=create_return_configured_target)370            stub.CreateAction = MagicMock(side_effect=create_return_action)371            stub.UpdateAction = MagicMock(side_effect=create_return_action)372            stub.UpdateConfiguredTarget = MagicMock(373                side_effect=create_return_configured_target)374            stub.UpdateTarget = MagicMock(side_effect=create_return_target)375            stub.FinalizeConfiguredTarget = MagicMock(376                side_effect=create_finalized_configured_target)377            stub.FinalizeTarget = MagicMock(378                side_effect=create_finalized_target)379            client = initialize_client(FLAGS)380            client.single_upload()381            stub.CreateConfiguration.assert_called_once()382            stub.CreateTarget.assert_called_once()383            stub.CreateConfiguredTarget.assert_called_once()384            stub.CreateAction.assert_called_once()385            stub.UpdateAction.assert_called_once()386            stub.UpdateConfiguredTarget.assert_called_once()387            stub.UpdateTarget.assert_called_once()388            stub.FinalizeConfiguredTarget.assert_called_once()389            stub.FinalizeTarget.assert_called_once()390class BatchUploadTest(absltest.TestCase):391    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)392    @patch(393        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'394    )395    def test_rpc_error(self, ResultStoreUploadStubMock, gen_new_uuid_mock):396        stub = ResultStoreUploadStubMock()397        stub.UploadBatch = MagicMock(return_value=DEFAULT_RPC_ERROR)398        client = initialize_client(FLAGS)399        upload_requests = client.create_resource_requests(400            TEST_INVOCATION_ID, TEST_CONFIG_ID)401        res = client.batch_upload(TEST_CURRENT_RESUME_TOKEN,402                                  TEST_NEXT_RESUME_TOKEN, TEST_INVOCATION_ID,403                                  upload_requests)404        self.assertEqual(res, DEFAULT_RPC_ERROR)405    @patch('resultstoreui_utils.gen_new_uuid', return_value=TEST_UUID)406    @patch(407        'resultstoreapi.cloud.devtools.resultstore_v2.proto.resultstore_upload_pb2_grpc.ResultStoreUploadStub'408    )409    def test_batch_upload_success(self, ResultStoreUploadStubMock,...

test_rclone.py

Source:test_rclone.py

...22        super(TestRclone, cls).tearDownClass()23    def test_rpc(self):24        o = self.rclone.rpc("rc/noop", a=42, b="string", c=[1234])25        self.assertEqual(dict(a=42, b="string", c=[1234]), o)26    def test_rpc_error(self):27        try:28            o = self.rclone.rpc("rc/error", a=42, b="string", c=[1234])29        except RcloneException as e:30            self.assertEqual(e.status, 500)31            self.assertTrue(e.output["error"].startswith("arbitrary error"))32        else:33            raise ValueError("Expecting exception")34if __name__ == '__main__':...

test_hello_world.py

Source:test_hello_world.py

2from src.shipchain_common.exceptions import RPCError3@pytest.fixture4def hw_string():  # This is only defined as a fixture as an example of fixtures in pytest5    return "Hello, world!"6def test_rpc_error(hw_string):7    with pytest.raises(RPCError, match=hw_string):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.