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

How to use test_sequence_number method in localstack

Best Python code snippet using localstack_python

test_amazon_kclpy_input_output_integration.py

Source:test_amazon_kclpy_input_output_integration.py

1import json2import re3from amazon_kclpy import kcl4from utils import make_io_obj5# Dummy record processor6class RecordProcessor(kcl.RecordProcessorBase):7    def __init__(self, expected_shard_id, expected_sequence_number):8        self.expected_shard_id = expected_shard_id9        self.expected_sequence_number = expected_sequence_number10        pass11    def initialize(self, shard_id):12        assert shard_id == self.expected_shard_id13        pass14    def process_records(self, records, checkpointer):15        seq = records[0].get('sequenceNumber')16        assert seq == self.expected_sequence_number17        try:18            checkpointer.checkpoint(seq)19            assert 0, "First checkpoint should fail"20        except Exception:21            # Try it one more time (this time it'll work)22            checkpointer.checkpoint(seq)23    def shutdown(self, checkpointer, reason):24        if 'TERMINATE' == reason:25            checkpointer.checkpoint()26'''27An input string which we'll feed to a file for kcl.py to read from.28'''29'''30This string is approximately what the output should look like. We remove whitespace when comparing this to what is31written to the outputfile.32'''33test_output_string = """34{"action": "status", "responseFor": "initialize"}35{"action": "checkpoint", "checkpoint": "456"}36{"action": "checkpoint", "checkpoint": "456"}37{"action": "status", "responseFor": "processRecords"}38{"action": "checkpoint", "checkpoint": null}39{"action": "status", "responseFor": "shutdown"}40"""41test_output_messages = [42    {"action": "status", "responseFor": "initialize"},43    {"action": "checkpoint", "sequenceNumber": "456", "subSequenceNumber": None},44    {"action": "checkpoint", "sequenceNumber": "456", "subSequenceNumber": None},45    {"action": "status", "responseFor": "processRecords"},46    {"action": "checkpoint", "sequenceNumber": None, "subSequenceNumber": None},47    {"action": "status", "responseFor": "shardEnded"}48]49def _strip_all_whitespace(s):50    return re.sub('\s*', '', s)51test_shard_id = "shardId-123"52test_sequence_number = "456"53test_input_messages = [54    {"action": "initialize", "shardId": test_shard_id, "sequenceNumber": test_sequence_number, "subSequenceNumber": 0},55    {"action": "processRecords", "millisBehindLatest": 1476889708000, "records":56        [57            {58                "action": "record", "data": "bWVvdw==", "partitionKey": "cat", "sequenceNumber": test_sequence_number,59                "subSequenceNumber": 0, "approximateArrivalTimestamp": 147688970700060            }61        ]62     },63    {"action": "checkpoint", "sequenceNumber": test_sequence_number, "subSequenceNumber": 0, "error": "Exception"},64    {"action": "checkpoint", "sequenceNumber": test_sequence_number, "subSequenceNumber": 0},65    {"action": "shardEnded"},66    {"action": "checkpoint", "sequenceNumber": test_sequence_number, "subSequenceNumber": 0}67]68def test_kcl_py_integration_test_perfect_input():69    test_input_json = "\n".join(map(lambda j: json.dumps(j), test_input_messages))70    input_file = make_io_obj(test_input_json)71    output_file = make_io_obj()72    error_file = make_io_obj()73    process = kcl.KCLProcess(RecordProcessor(test_shard_id, test_sequence_number),74                             input_file=input_file, output_file=output_file, error_file=error_file)75    process.run()76    '''77    The strings are approximately the same, modulo whitespace.78    '''79    output_message_list = filter(lambda s: s != "", output_file.getvalue().split("\n"))80    responses = [json.loads(s) for s in output_message_list]81    assert len(responses) == len(test_output_messages)82    for i in range(len(responses)):83        assert responses[i] == test_output_messages[i]84    '''85    There should be some error output but it seems like overly specific to make sure that a particular message is printed.86    '''87    error_output = error_file.getvalue()...

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.