KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use put_record_batch method in localstack

Best Python code snippet using localstack_python

test_firehose_dal.py

Source:test_firehose_dal.py

...29    firehose_service.update_failed_by_error_code(kinesis_response)30    assert firehose_service.failed_by_error_code == {RANDOM_ERROR_CODE: 1}31def test_put_records_happy_flow():32    firehose = FirehoseDal(stream_name="stream_name", account_id="random")33    assert firehose.put_record_batch(records=[{"id": 1}]) == 134def test_put_records_happy_flow_with_non_default_batch_size():35    firehose = FirehoseDal(stream_name="stream_name", batch_size=1, account_id="random")36    assert firehose.put_record_batch(records=[{"id": 1}]) == 137def test_put_records_with_record_too_big():38    firehose = FirehoseDal(stream_name="stream_name", account_id="random")39    assert firehose.put_record_batch(records=[{"id": ("x" * 2048000)}]) == 040def test_put_records_with_record_invalid():41    firehose = FirehoseDal(stream_name="stream_name", account_id="random")42    assert firehose.put_record_batch(records=[Exception()]) == 0  # noqa43def test_put_records_error_on_first_try_success_in_second_try(monkeypatch):44    monkeypatch.setattr(45        FirehoseDal,46        "get_boto_client",47        lambda x, y: MockFirehoseBotoClientRetryWithError(),48    )49    firehose = FirehoseDal(stream_name="stream_name", account_id="random")50    assert firehose.put_record_batch(records=[{"id": 1}]) == 151def test_put_records_exception_on_first_try_success_in_second_try(monkeypatch):52    client = MockFirehoseBotoClientException()53    monkeypatch.setattr(FirehoseDal, "get_boto_client", lambda x, y: client)54    firehose = FirehoseDal(stream_name="stream_name", account_id="random")55    assert firehose.put_record_batch(records=[{"id": 1}]) == 156def test_put_records_dont_retry_to_many_times(monkeypatch):57    monkeypatch.setattr(58        FirehoseDal,59        "get_boto_client",60        lambda x, y: MockFirehoseBotoClientRetryWithError(3),61    )62    firehose = FirehoseDal(63        stream_name="stream_name", max_retry_count=2, account_id="random"64    )65    assert firehose.put_record_batch(records=[{"id": 1}]) == 066@pytest.mark.skip_firehose_dal_mock67@mock_sts68def test_get_boto_client_in_china_should_connect_to_global(monkeypatch):69    monkeypatch.setenv("LUMIGO_LOGS_EDGE_AWS_ACCESS_KEY_ID", "key1")70    monkeypatch.setenv("LUMIGO_LOGS_EDGE_AWS_SECRET_ACCESS_KEY", "secret1")71    monkeypatch.setenv("AWS_REGION", "cn-northwest-1")72    client = FirehoseDal.get_boto_client("111111111111", "unittests")73    assert client.meta.region_name == "us-west-2"74class MockFirehoseBotoClientRetryWithException:75    retry = 176    def put_record_batch(self, DeliveryStreamName: str, Records: List[dict]):77        if self.retry > 0:78            self.retry -= 179            raise Exception()80        return {81            "FailedPutCount": 0,82            "RequestResponses": [{"RecordId": "1"} for _ in Records],83        }84class MockFirehoseBotoClientRetryWithError:85    def __init__(self, retry=1):86        self.retry = retry87    def put_record_batch(self, DeliveryStreamName: str, Records: List[dict]):88        if self.retry > 0:89            self.retry -= 190            return {91                "FailedPutCount": 0,92                "RequestResponses": [93                    {94                        "ErrorCode": "ServiceUnavailableException",95                        "ErrorMessage": "ServiceUnavailableException",96                    }97                    for _ in Records98                ],99            }100        return {101            "FailedPutCount": 0,102            "RequestResponses": [{"RecordId": "1"} for _ in Records],103        }104class MockFirehoseBotoClientException:105    retry = 1106    records: List = []107    def put_record_batch(self, DeliveryStreamName: str, Records: List[dict]):108        if self.retry > 0:109            self.retry -= 1110            raise Exception()111        self.records.extend(Records)112        return {113            "FailedPutCount": 0,114            "RequestResponses": [{"RecordId": "1"} for _ in Records],115        }116@pytest.fixture117def kinesis_success_item() -> dict:118    return {"RecordId": RANDOM_RECORD_ID}119@pytest.fixture120def kinesis_failed_item() -> dict:121    return {"ErrorCode": RANDOM_ERROR_CODE}

lambdacode.py

Source:lambdacode.py

...44        }45        records.append(logEvent)46        if len(records) > 499:47            print("Records are greater than 499 lets batch and send to firehose %s " % len(records))48            response = firehose.put_record_batch(49                DeliveryStreamName = os.environ['DELIVERY_STREAM_NAME'],50                Records = records51            )52            print("Response from firehose put_record_batch in message process loop is: %s" % response)53            records = []54    if len(records) > 0:55        response = firehose.put_record_batch(56            DeliveryStreamName = os.environ['DELIVERY_STREAM_NAME'],57            Records = records58        )59        print("Response from firehose put_record_batch after message process loop is: %s" % response)60def getEniId(message):61    records = message.split()62    if len(records) > 0:63        return records[2]64    else:65        return None66def getSgIds(netInfObj):67    sgroups = netInfObj.groups68    if len(sgroups) > 0:69        tempsgroupids = []...

data_lake.py

Source:data_lake.py

...42                raise e43            else:44                warn(e)45                return False, str(e)46    def put_record_batch(self, items: list) -> Tuple[bool, str]:47        try:48            response = self.client.put_record_batch(49                DeliveryStreamName=self.DELIVERY_STREAM_NAME,50                Records=[{'Data': self.convert_to_byte(item)} for item in items],51            )52            return True, None53        except Exception as e:54            if settings.DEBUG:55                raise e56            else:57                warn(e)...

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.