Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use delete_stream method in localstack

Best Python code snippet using localstack_python

test_harvest_start_date.py

Source:test_harvest_start_date.py

...34                                "project_users": [], "user_roles": []}35        # Protect against surviving projects from previous failures36        project_1 = ""37        for project in get_all('projects'):38            delete_stream('projects', project['id'])39        # Create dummy data in specifc streams prior to first sync to ensure they are captured        40        for itter in range(2):41            logging.info("Creating {} round(s) of data ...".format(itter + 1))42            # Clients43            client_1 = create_client()44            cls._teardown_delete["clients"].append({"id": client_1["id"]})45            # Contacts46            contact_1 = create_contact(client_1['id'])47            cls._teardown_delete["contacts"].append({"id": contact_1["id"]})48            # Projects49            if itter < 2: # Protect against project max (2)50                project_1 = create_project(client_1['id'])51                cls._teardown_delete["projects"].append({"id": project_1["id"]})52            project_user_1 = create_project_user(project_1['id'], get_random("users"))53            cls._teardown_delete["project_users"].append({"id": project_user_1["id"]})54            55            # Tasks56            task_1 = create_task()57            cls._teardown_delete["tasks"].append({"id": task_1["id"]})58            project_task_1 = create_project_task(project_1['id'], task_1['id'])59            cls._teardown_delete["project_tasks"].append({"id": project_task_1["id"]})60            # Users ( This is necessary to ensure user_projects data exists)61            user_1 = update_user(get_random('users'))62            # Roles63            role_1 = create_role()64            cls._teardown_delete["roles"].append({"id": role_1["id"]})65            user_role_1 = update_role(role_1['id'], [get_random("users")])66            cls._teardown_delete["user_roles"].append({"id": user_role_1["id"]})67            68            # Estimates69            estimate_1 = create_estimate(client_1['id'])70            cls._teardown_delete["estimates"].append({"id": estimate_1['id']})71            estimate_message_1 = create_estimate_message(estimate_1['id'])72            cls._teardown_delete["estimate_messages"].append({"id": estimate_message_1['id']})73            estimate_item_category_1 = create_estimate_item_category()74            cls._teardown_delete["estimate_item_categories"].append({"id": estimate_item_category_1['id']})75            # Invoices76            invoice_1 = create_invoice(client_id=client_1['id'], project_id=project_1['id'])77            cls._teardown_delete["invoices"].append({"id": invoice_1["id"]})78            invoice_message_1 = create_invoice_message(invoice_1['id'])79            cls._teardown_delete["invoice_messages"].append({"id": invoice_message_1["id"]})80            invoice_payment_1 = create_invoice_payment(invoice_1['id'])81            cls._teardown_delete["invoice_payments"].append({"id": invoice_payment_1["id"]})82            invoice_item_category_1 = create_invoice_item_category()83            cls._teardown_delete["invoice_item_categories"].append({"id": invoice_item_category_1["id"]})84            # Time Entries85            time_entry_1 = create_time_entry(project_1['id'], task_1['id'])86            cls._teardown_delete["time_entries"].append({"id": time_entry_1["id"]})87            # Expesnes88            expense_category_1 = create_expense_category()89            cls._teardown_delete["expense_categories"].append({"id": expense_category_1['id']})90            expense_1 = create_expense(project_1['id'])91            cls._teardown_delete["expenses"].append({"id": expense_1['id']})92    @classmethod93    def tearDownClass(cls):94        # Clean up any data created in the setup95        logging.info("Start Teardown")96        # Projects97        for project in cls._teardown_delete['projects']:98            delete_stream('projects', project['id'])99        # Estimates100        for estimate in cls._teardown_delete['estimates']:101            delete_stream('estimates', estimate['id'])102        for category in cls._teardown_delete['estimate_item_categories']:103            delete_stream('estimate_item_categories', category['id'])104        # Time Entries TODO fix the delete methods105        # for time_entry in cls._teardown_delete['time_entries']:106        #     delete_stream('time_entries', time_entry['id'])107        # Tasks108        # for task in cls._teardown_delete['tasks']:109        #     delete_stream('tasks', task['id'])110        # Invoices111        for invoice in cls._teardown_delete['invoices']:112            delete_stream('invoices', invoice['id'])113        for category in cls._teardown_delete['invoice_item_categories']:114            delete_stream('invoice_item_categories', category['id'])115        # Clients116        for client in cls._teardown_delete['clients']:117            delete_stream('clients', client['id'])118        # Expenses119        # for expense in cls._teardown_delete['expenses']:120        #     delete_stream('expenses', expense['id'])121        # for expense_category in cls._teardown_delete['expense_categories']:122        #     delete_stream('expense_categories', expense_category['id'])123        for role in cls._teardown_delete['roles']:124            delete_stream('roles', role['id'])125        ##########  Uncomment for PURGE MODE ###############126        # stream = ""127        # logging.info("Comencing PURGE of stream: {}".format(stream))128        # all_of_stream = get_all(stream)129        # while all_of_stream:130        #     for s in get_all(stream):131        #         try:132        #             delete_stream(stream, s['id'])133        #         except AssertionError:134        #             continue135        #     all_of_stream = get_all(stream)136        ####################################################137    def name(self):138        return "tap_tester_harvest_start_date"139    def do_test(self, conn_id):140        """Test we get a lot of data back based on the start date configured in base"""141        # Select all streams and all fields within streams142        found_catalogs = menagerie.get_catalogs(conn_id)143        incremental_streams = {key for key, value in self.expected_replication_method().items()144                               if value == self.INCREMENTAL}145        # IF THERE ARE STREAMS THAT SHOULD NOT BE TESTED146        # REPLACE THE EMPTY SET BELOW WITH THOSE STREAM...

delete_stream.py

Source:delete_stream.py

2import os3import boto3456def delete_stream(StreamName=None, region_name='us-west-2'):7    """8    Executes Kinesis.Client.delete_stream function9    """10    assert StreamName is not None1112    client = boto3.client('kinesis', region_name=region_name)13    return client.delete_stream(14        StreamName=StreamName15    )161718def parse_known_args():19    # AWS credentials should be provided as environ variables20    if 'AWS_ACCESS_KEY_ID' not in os.environ:21        print('Error. Please setup AWS_ACCESS_KEY_ID')22        exit(1)2324    elif 'AWS_SECRET_ACCESS_KEY' not in os.environ:25        print('Error. Please setup AWS_SECRET_ACCESS_KEY')26        exit(1)2728    parser = argparse.ArgumentParser()29    parser.add_argument(30        '--region_name', required=False,31        help='AWS Region', default='us-west-2')32    parser.add_argument('--StreamName', required=True, help='Stream name')33    args, extra_params = parser.parse_known_args()34    return args, extra_params353637def main():38    """39    Deletes an Amazon Kinesis stream using boto3 library40    """4142    args, _ = parse_known_args()43    print(44        delete_stream(45            StreamName=args.StreamName, region_name=args.region_name))464748if __name__ == '__main__':
...

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.