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

How to use _put_records method in localstack

Best Python code snippet using localstack_python

test_all.py

Source:test_all.py

...23    """Asserts that a non-critical error is routed to the error stream."""24    # Get the shard iterators *before* you put records at the input25    shard_iterators = get_shard_iterators(kinesis, ostream)26    error_shard_iterators = get_shard_iterators(kinesis, estream)27    _put_records(kinesis, istream, recs)28    nbrecs = len(recs)29    retrieved_recs = _retrieve_records(2*(nbrecs-1), kinesis, shard_iterators)30    assert len(retrieved_recs) == 2*(nbrecs-1)31    retrieved_error_recs = _retrieve_records(1, kinesis, error_shard_iterators)32    assert len(retrieved_error_recs) == 133def test_io_streams_put_get_record(34        kinesis, input_stream_name, output_stream_name):35    """Put records at the input, then read them from the output."""36    sample_records = make_records(2)37    # Get the shard iterators *before* you put records at the input38    shard_iterators = get_shard_iterators(kinesis, output_stream_name)39    _put_records(kinesis, input_stream_name, sample_records)40    nbrecs = len(sample_records)41    retrieved_recs = _retrieve_records(2*nbrecs, kinesis, shard_iterators)42    # Two pipelines that produce one event for each input event43    assert len(retrieved_recs) == 2*nbrecs44    retrieved_recs = [json.loads(x["Data"].decode()) for x in retrieved_recs]45    retrieved_ids = {x["id"] for x in retrieved_recs}46    put_ids = {x['id'] for x in sample_records}47    assert not retrieved_ids.difference(put_ids)48    assert all("input_filter" not in ev and "input_mapper" in ev and49               "received_at" in ev for ev in retrieved_recs)50    assert all("batch_mapped" in ev for ev in retrieved_recs)51def test_set_get_state(52        kinesis, input_stream_name, output_stream_name):53    """Put and read a record from the input stream."""54    sample_records = make_records(2)55    nbrecs = len(sample_records)56    # Get the shard iterators *before* you put the records at the input57    shard_iterators = get_shard_iterators(kinesis, output_stream_name)58    # Put the same record multiple times in the stream59    _put_records(kinesis, input_stream_name,60                 [sample_records[0] for _ in range(nbrecs)])61    retrieved_recs = _retrieve_records(nbrecs, kinesis, shard_iterators)62    # The same input event should be sent to the two output streams63    assert len(retrieved_recs) == 264    retrieved_event = json.loads(retrieved_recs[0]["Data"].decode())65    retrieved_id = retrieved_event["id"]66    put_id = sample_records[0]["id"]67    assert put_id == retrieved_id68    assert "input_filter" not in retrieved_event \69        and "input_mapper" in retrieved_event \70        and "received_at" in retrieved_event71def _put_records(kinesis, input_stream_name, records):72    """Put records in the input Kinesis stream."""73    response = kinesis.put_records(74        StreamName=input_stream_name,75        Records=[76            {77                "Data": json.dumps(rec),78                "PartitionKey": str(uuid.uuid4())79            } for rec in records])80    assert response["ResponseMetadata"]["HTTPStatusCode"] == 20081def _retrieve_records(nbrecs, kinesis, shard_iterators):82    """Retrieve records from the output Kinesis stream."""83    retrieved_recs = []84    # Just some rule-of-thumb timeout85    timeout = min(max(20, 5 * nbrecs), 120)...

enrich-stream.py

Source:enrich-stream.py

...32    context.logger.info_with('Incoming message', msg=msg)33    enrichment_data = _search_kv(msg, config)34    context.logger.info_with('Enrichment data', enrichment_data=enrichment_data)35    msg['enrichment'] = enrichment_data36    _put_records([msg], config)37    context.logger.debug_with('Output message', msg=msg)38def _get_url(v3io_api, container_name, collection_path):39    return f'http://{v3io_api}/{container_name}/{collection_path}'40def _get_headers(v3io_function, v3io_access_key):41    return {42        'Content-Type': 'application/json',43        'X-v3io-function': v3io_function,44        'cache-control': 'no-cache',45        'x-v3io-session-key': v3io_access_key46    }47def _search_kv(msg, config):48    v3io_api = config['v3io_api']49    v3io_username = config['v3io_username']50    container_name = config['container_name']51    search_value = msg[config['input_stream_search_key']]52    table_path_and_key = f"{v3io_username}/examples/stream-enrich/{config['table_name']}/{search_value}"53    v3io_access_key = config['v3io_access_key']54    url = _get_url(v3io_api, container_name, table_path_and_key)55    headers = _get_headers('GetItem', v3io_access_key)56    resp = requests.request('POST', url, json={}, headers=headers)57    json_response = json.loads(resp.text)58    response = {}59    if 'Item' in json_response:60        response = json_response['Item']61    return response62def _put_records(items, config):63    v3io_api = config['v3io_api']64    v3io_username = config['v3io_username']65    container_name = config['container_name']66    output_stream_path = f"{v3io_username}/examples/stream-enrich/{config['output_stream_name']}/"67    v3io_access_key = config['v3io_access_key']68    records = _items_to_records(items)69    url = _get_url(v3io_api, container_name, output_stream_path)70    headers = _get_headers('PutRecords', v3io_access_key)71    return requests.request('PUT', url, json=records, headers=headers)72def _item_to_b64(item):73    item_string = json.dumps(item)74    return base64.b64encode(item_string.encode('utf-8')).decode('utf-8')75def _items_to_records(items):76    return {'Records': [{'Data': _item_to_b64(item)} for item in items]}

test_deletion_check_initiator.py

Source:test_deletion_check_initiator.py

...17    yield table_name18    mock_dynamo_client.delete_table(TableName=table_name)19@pytest.fixture(scope="function")20def put_records(mock_dynamo_client, test_dynamo_table):21    def _put_records(records):22        for record in records:23            mock_dynamo_client.put_item(24                TableName=test_dynamo_table, Item={"id": {"S": record["id"]}}25            )26    yield _put_records27@pytest.fixture(scope="function")28def deletion_check_initiator(29    mock_dynamo_client, mock_sns_client, test_topic_arn, test_dynamo_table30):31    yield DeletionCheckInitiator(32        dynamo_client=mock_dynamo_client,33        sns_client=mock_sns_client,34        reindexer_topic_arn=test_topic_arn,35        source_table_name=test_dynamo_table,...

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.