How to use the pysoa.test.compatibility.mock.patch function in pysoa

    @mock.patch('pysoa.common.transport.redis_gateway.core.SentinelRedisClient')
    @mock.patch('pysoa.common.transport.redis_gateway.core.StandardRedisClient')
    def test_chunking_supported_on_server(self, mock_standard, mock_sentinel):
        # noinspection PyArgumentList
        core = RedisTransportServerCore(
            backend_type=REDIS_BACKEND_TYPE_SENTINEL,
            backend_layer_kwargs={
                'connection_kwargs': {'hello': 'world'},
                'hosts': [('another_host', 6379)],
                'redis_db': 5,
                'redis_port': 1098,
                'sentinel_refresh_interval': 13,
                'sentinel_services': ['svc1', 'svc2', 'svc3'],
                'sentinel_failover_retries': 5,
            },
            message_expiry_in_seconds=45,
            queue_capacity=7500,

    @mock.patch('pysoa.common.transport.redis_gateway.core.StandardRedisClient')
    def test_receive_chunking_fails_on_client_missing_header(self, mock_standard):
        core = self._get_client_core(receive_timeout_in_seconds=3, message_expiry_in_seconds=10)

        message = {'request_id': 79, 'meta': {'yes': 'no'}, 'body': {'baz': 'qux'}}
        message['body'].update({'key-{}'.format(i): 'value-{}'.format(i) for i in range(200)})  # type: ignore

        serialized = MsgpackSerializer().dict_to_blob(message)

        mock_standard.return_value.get_connection.return_value.blpop.side_effect = [
            [True, (b'pysoa-redis/3//chunk-count:4;' + serialized[0:1000])],
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:2;' + serialized[1000:2000])],
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:3;' + serialized[2000:3000])],
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:4;' + serialized[3000:])],
        ]

        with pytest.raises(InvalidMessageError) as error_context:

    @mock.patch('pysoa.common.transport.redis_gateway.core.SentinelRedisClient')
    @mock.patch('pysoa.common.transport.redis_gateway.core.StandardRedisClient')
    def test_chunking_not_supported_on_client(self, mock_standard, mock_sentinel):
        with pytest.raises(TypeError) as error_context:
            # noinspection PyArgumentList
            RedisTransportClientCore(  # type: ignore
                backend_type=REDIS_BACKEND_TYPE_SENTINEL,
                backend_layer_kwargs={
                    'connection_kwargs': {'hello': 'world'},
                    'hosts': [('another_host', 6379)],
                    'redis_db': 5,
                    'redis_port': 1098,
                    'sentinel_refresh_interval': 13,
                    'sentinel_services': ['svc1', 'svc2', 'svc3'],
                    'sentinel_failover_retries': 5,
                },
                message_expiry_in_seconds=45,

def test_call_actions_raises_exception_on_job_error(self):
        """Client.call_actions raises Client.JobError when a JobError occurs on the server."""
        errors = [Error(code=ERROR_CODE_SERVER_ERROR, message='Something went wrong!')]
        with mock.patch(
            'pysoa.server.server.Server.execute_job',
            new=mock.Mock(side_effect=JobError(errors)),
        ):
            client = Client(self.client_settings)
            with self.assertRaises(Client.JobError) as e:
                client.call_action(SERVICE_NAME, 'action_1')
            self.assertEqual(e.exception.errors, errors)

    @mock.patch('redis.sentinel.Sentinel')
    def test_master_not_found_no_retry(self, mock_sentinel):
        mock_sentinel.return_value.master_for.return_value = MockSentinelRedis()

        client = self._set_up_client(sentinel_services=['service1', 'service2', 'service3'])
        client.reset_clients()

        mock_sentinel.return_value.master_for.reset_mock()
        mock_sentinel.return_value.master_for.side_effect = redis.sentinel.MasterNotFoundError

        with self.assertRaises(CannotGetConnectionError):
            client.get_connection('test_master_not_found_no_retry')

        self.assertEqual(1, mock_sentinel.return_value.master_for.call_count)
        self.assertIn(mock_sentinel.return_value.master_for.call_args[0][0], {'service1', 'service2', 'service3'})

    @mock.patch('pysoa.common.transport.redis_gateway.core.StandardRedisClient')
    def test_send_chunking_works_three_chunks(self, mock_standard):
        core = self._get_server_core(
            chunk_messages_larger_than_bytes=102400,
            maximum_message_size_in_bytes=102400 * 6,
        )

        meta = {'protocol_version': ProtocolVersion.VERSION_3}
        body = {'test': ['payload%i' % i for i in range(10000, 30000)]}  # 2.5 chunks needed

        core.send_message('test_send_chunking_works_three_chunks', 103, meta, body)

        assert mock_standard.return_value.send_message_to_queue.call_count == 3

        starts_with = b'pysoa-redis/3//content-type:application/msgpack;chunk-count:3;chunk-id:'
        starts_with_length = len(starts_with) + 2

    @mock.patch('pysoa.common.transport.redis_gateway.core.StandardRedisClient')
    def test_receive_chunking_prohibited_on_server(self, mock_standard):
        core = self._get_server_core(receive_timeout_in_seconds=3, message_expiry_in_seconds=10)

        message = {'request_id': 79, 'meta': {'yes': 'no'}, 'body': {'baz': 'qux'}}
        message['body'].update({'key-{}'.format(i): 'value-{}'.format(i) for i in range(200)})  # type: ignore

        serialized = MsgpackSerializer().dict_to_blob(message)

        mock_standard.return_value.get_connection.return_value.blpop.side_effect = [
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:1;' + serialized[0:1000])],
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:2;' + serialized[1000:2000])],
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:3;' + serialized[2000:3000])],
            [True, (b'pysoa-redis/3//chunk-count:4;chunk-id:4;' + serialized[3000:])],
        ]

        with pytest.raises(InvalidMessageError) as error_context:

    @mock.patch('redis.sentinel.Sentinel', new=MockSentinel)
    def test_services_send_receive(self):
        client = self._set_up_client(sentinel_services=['service1', 'service2', 'service3'])

        payload = {'test': 'test_services_send_receive'}

        client.send_message_to_queue(
            queue_key='test_services_send_receive',
            message=msgpack.packb(payload, use_bin_type=True),
            expiry=10,
            capacity=10,
            connection=client.get_connection('test_services_send_receive'),
        )

        message = None
        for i in range(3):
            # Message will be on random server

    @mock.patch('pysoa.server.autoreload.get_reloader')
    def test_only_file_watcher_argument_some_values(self, mock_get_reloader):
        server_getter = mock.MagicMock()

        sys.argv = ['/path/to/example_service/standalone.py', '--use-file-watcher', 'example,pysoa,conformity']

        standalone.simple_main(server_getter)  # type: ignore

        server_getter.assert_called_once_with()
        self.assertFalse(server_getter.return_value.main.called)

        assert mock_get_reloader.call_count == 1
        assert mock_get_reloader.call_args_list[0][0][0] in ('', 'pytest', 'coverage')
        assert mock_get_reloader.call_args_list[0][0][1] == ['example', 'pysoa', 'conformity']
        assert mock_get_reloader.call_args_list[0][1]['signal_forks'] is False

        self.assertEqual(1, mock_get_reloader.return_value.main.call_count)