Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use delete_table method in localstack

Best Python code snippet using localstack_python

test_datastore.py

Source:test_datastore.py

...65        received = self._ds.init_table(self._image_table_name)66        msg = 'Table initialisation (existing table) should return False'67        self.assertFalse(received, msg)68        # Now delete.69        received = self._ds.delete_table(self._image_table_name)70        msg = 'Table deletion (existing table) should return True'71        self.assertTrue(received, msg)72        # Try delete again.  Missing table should fail.73        received = self._ds.delete_table(self._image_table_name)74        msg = 'Table deletion (missing table) should return False'75        self.assertFalse(received, msg)76    def test_create_write_no_connection(self):77        """Create an Accumulo writer object: no connection.78        """79        received = self._ds._create_writer(table='dodgy')80        msg = 'Accumulo writer object (no connection) not None'81        self.assertIsNone(received, msg)82    def test_ingest_no_connection(self):83        """Accumulo ingest: no connection.84        """85        from geoutils.tests.files.ingest_data_01 import DATA86        received = self._ds.ingest(DATA)87        msg = 'Accumulo ingest (no connection) not False'88        self.assertFalse(received, msg)89    def test_ingest_no_row_id(self):90        """Ingest attempt with no datastore connection.91        """92        data = {}93        received = self._ds.ingest(data)94        msg = 'Ingest status with no row_id not False'95        self.assertFalse(received, msg)96    def test_ingest(self):97        """Ingest the metadata component into the datastore.98        """99        image_stream_file = os.path.join('geoutils',100                                         'tests',101                                         'files',102                                         'image_stream.out')103        image_fh = open(image_stream_file, 'rb')104        thumb_stream_file = os.path.join('geoutils',105                                         'tests',106                                         'files',107                                         '300x300_stream.out')108        thumb_fh = open(thumb_stream_file, 'rb')109        from geoutils.tests.files.ingest_data_01 import DATA110        image_tbl = self._image_table_name111        thumb_tbl = self._thumb_table_name112        table = DATA['tables']113        table[image_tbl] = {'cf': {'cq': {'x_coord_size': '1024',114                                          'y_coord_size': '1024'},115                                   'val': {'image': image_fh.read}}}116        table[thumb_tbl] = {'cf': {'cq': {'x_coord_size': '300',117                                          'y_coord_size': '300'},118                                   'val': {'image': thumb_fh.read}}}119        self._ds.connect()120        self._ds.init_table(self._meta_table_name)121        self._ds.init_table(self._image_table_name)122        self._ds.init_table(self._thumb_table_name)123        received = self._ds.ingest(DATA)124        msg = 'Ingest status with datastore connection not True'125        self.assertTrue(received, msg)126        # Clean up.127        DATA['tables'].pop(self._image_table_name, None)128        DATA['tables'].pop(self._thumb_table_name, None)129        image_fh.close()130        self._ds.delete_table(self._meta_table_name)131        self._ds.delete_table(self._image_table_name)132        self._ds.delete_table(self._thumb_table_name)133    def test_ingest_spatial_index(self):134        """Ingest the metadata component: spatial index.135        """136        from geoutils.tests.files.ingest_data_01 import DATA137        self._ds.connect()138        self._ds.init_table(self._image_spatial_index_table_name)139        received = self._ds.ingest(DATA)140        msg = 'Ingest status with datastore connection not True'141        self.assertTrue(received, msg)142        # Clean up.143        self._ds.delete_table(self._image_spatial_index_table_name)144    def test_ingest_with_metasearch(self):145        """Ingest the metadata component: with metasearch.146        """147        image_stream_file = os.path.join('geoutils',148                                         'tests',149                                         'files',150                                         'image_stream.out')151        image_fh = open(image_stream_file, 'rb')152        thumb_stream_file = os.path.join('geoutils',153                                         'tests',154                                         'files',155                                         '300x300_stream.out')156        thumb_fh = open(thumb_stream_file, 'rb')157        from geoutils.tests.files.ingest_data_02 import DATA158        image_tbl = self._image_table_name159        thumb_tbl = self._thumb_table_name160        table = DATA['tables']161        table[image_tbl] = {'cf': {'cq': {'x_coord_size': '1024',162                                          'y_coord_size': '1024'},163                                   'val': {'image': image_fh.read}}}164        table[thumb_tbl] = {'cf': {'cq': {'x_coord_size': '300',165                                          'y_coord_size': '300'},166                                   'val': {'image': thumb_fh.read}}}167        self._ds.connect()168        self._ds.init_table(self._meta_table_name)169        self._ds.init_table(self._metasearch_table_name)170        self._ds.init_table(self._image_table_name)171        self._ds.init_table(self._thumb_table_name)172        received = self._ds.ingest(DATA)173        msg = 'Ingest status with datastore connection not True'174        self.assertTrue(received, msg)175        # Clean up.176        DATA['tables'].pop(self._image_table_name, None)177        DATA['tables'].pop(self._thumb_table_name, None)178        image_fh.close()179        self._ds.delete_table(self._meta_table_name)180        self._ds.delete_table(self._metasearch_table_name)181        self._ds.delete_table(self._image_table_name)182        self._ds.delete_table(self._thumb_table_name)183    def test_ingest_val_column_not_callable(self):184        """Ingest: value column not a callable.185        """186        from geoutils.tests.files.ingest_data_01 import DATA187        thumb_tbl = self._thumb_table_name188        table = DATA['tables']189        table[thumb_tbl] = {'cf': {'cq': {'x_coord_size': '300',190                                          'y_coord_size': '300'},191                                   'val': {'image': 'just a scalar'}}}192        self._ds.connect()193        self._ds.init_table(self._meta_table_name)194        self._ds.init_table(self._thumb_table_name)195        received = self._ds.ingest(DATA)196        msg = 'Ingest with value column as a scalar not True'197        self.assertTrue(received, msg)198        # Clean up.199        DATA['tables'].pop(self._thumb_table_name, None)200        self._ds.delete_table(self._meta_table_name)201        self._ds.delete_table(self._thumb_table_name)202    def test_ingest_from_file(self):203        """Attempt to ingest from NITF file.204        """205        ntf_file = os.path.join('geoutils',206                                'tests',207                                'files',208                                'i_3001a.ntf')209        self._ds.connect()210        self._ds.init_table(self._meta_table_name)211        self._ds.init_table(self._image_table_name)212        self._ds.init_table(self._thumb_table_name)213        standard = geoutils.Standard(source_filename=ntf_file)214        standard.open()215        self._ds.ingest(standard(dry=True))216        # If you want to ingest some sample data into the proxy server217        # and block (so that you can connect via the client) then218        # uncomment the following two lines.219        # import time220        # time.sleep(1000)221        # Clean up.222        self._ds.delete_table(self._meta_table_name)223        self._ds.delete_table(self._image_table_name)224        self._ds.delete_table(self._thumb_table_name)225    def test_ingest_from_file_document_partioned_index(self):226        """Attempt to ingest from NITF file.227        """228        ntf_file = os.path.join('geoutils',229                                'tests',230                                'files',231                                'i_3001a.ntf')232        self._ds.connect()233        self._ds.init_table(self._meta_table_name)234        self._ds.init_table(self._metasearch_table_name)235        self._ds.init_table(self._image_table_name)236        self._ds.init_table(self._thumb_table_name)237        standard = geoutils.Standard(source_filename=ntf_file)238        standard.open()239        self._ds.ingest(standard(dry=True))240        # If you want to ingest some sample data into the proxy server241        # and block (so that you can connect via the client) then242        # uncomment the following two lines.243        # import time244        # time.sleep(1000)245        # Clean up.246        self._ds.delete_table(self._meta_table_name)247        self._ds.delete_table(self._metasearch_table_name)248        self._ds.delete_table(self._image_table_name)249        self._ds.delete_table(self._thumb_table_name)250    def test_ingest_from_file_spatial_index(self):251        """Attempt to ingest from NITF file.252        """253        ntf_file = os.path.join('geoutils',254                                'tests',255                                'files',256                                'i_3001a.ntf')257        self._ds.connect()258        self._ds.init_table(self._meta_table_name)259        self._ds.init_table(self._image_spatial_index_table_name)260        self._ds.init_table(self._metasearch_table_name)261        self._ds.init_table(self._image_table_name)262        self._ds.init_table(self._thumb_table_name)263        standard = geoutils.Standard(source_filename=ntf_file)264        standard.open()265        self._ds.ingest(standard(dry=True))266        # If you want to ingest some sample data into the proxy server267        # and block (so that you can connect via the client) then268        # uncomment the following two lines.269        # import time270        # time.sleep(1000)271        # Clean up.272        self._ds.delete_table(self._meta_table_name)273        self._ds.delete_table(self._image_spatial_index_table_name)274        self._ds.delete_table(self._metasearch_table_name)275        self._ds.delete_table(self._image_table_name)276        self._ds.delete_table(self._thumb_table_name)277#    def test_ingest_multiband_image(self):278#        """Attempt to ingest from multiband (RGB) NITF file.279#        """280#        mb_file = '11OCT12192723-S2AS-052799693030_01_P001.NTF'281#        ntf_file = os.path.join(os.sep, 'media', 'sf_dev', mb_file)282#283#        self._ds.connect()284#        self._ds.init_table(self._meta_table_name)285#        self._ds.init_table(self._image_spatial_index_table_name)286#        self._ds.init_table(self._metasearch_table_name)287#        self._ds.init_table(self._image_table_name)288#        self._ds.init_table(self._thumb_table_name)289#290#        standard = geoutils.Standard(source_filename=ntf_file)291#        standard.open()292#        self._ds.ingest(standard(dry=True))293#294#        # If you want to ingest some sample data into the proxy server295#        # and block (so that you can connect via the client) then296#        # uncomment the following two lines.297#        # import time298#        # time.sleep(1000)299#300#        # Clean up.301#        self._ds.delete_table(self._meta_table_name)302#        self._ds.delete_table(self._image_spatial_index_table_name)303#        self._ds.delete_table(self._metasearch_table_name)304#        self._ds.delete_table(self._image_table_name)305#        self._ds.delete_table(self._thumb_table_name)306    def test_delete_table_no_connection(self):307        """Delete an Accumulo table: no connection.308        """309        received = self._ds.delete_table(name='dodge')310        msg = 'Accumulo table deletion (no connection) not False'311        self.assertFalse(received, msg)312    def test_exists_table_no_connection(self):313        """Table exists check: no connection.314        """315        received = self._ds.exists_table(name='dodge')316        msg = 'Table exists check (no connection) is not False'317        self.assertFalse(received, msg)318    def test_exists_table_no_table(self):319        """Table exists check: no table.320        """321        self._ds.connect()322        received = self._ds.exists_table(name='dodge')323        msg = 'Table exists check (no table) is not False'324        self.assertFalse(received, msg)325    def test_exists_table(self):326        """Table exists check.327        """328        self._ds.connect()329        self._ds.init_table(self._meta_table_name)330        received = self._ds.exists_table(name=self._meta_table_name)331        msg = 'Table exists check is not True'332        self.assertTrue(received, msg)333        # Clean up.334        self._ds.delete_table(self._meta_table_name)335    def test_index(self):336        """Index creating mutation.337        """338        pass339    @classmethod340    def tearDownClass(cls):341        """Shutdown the Accumulo mock proxy server (if enabled)342        """343        cls._mock.stop()344        del cls._meta_table_name345        del cls._image_spatial_index_table_name346        del cls._metasearch_table_name347        del cls._image_table_name348        del cls._thumb_table_name...

db_mysql.py

Source:db_mysql.py

...45        db.execute_deferred_sql()46        constraint = db._find_foreign_constraints(main_table, 'foreign_id')[0]47        references = db._lookup_constraint_references(main_table, constraint)48        self.assertEquals((reference_table, 'id'), references)49        db.delete_table(main_table)50        db.delete_table(reference_table)51    def test_reverse_column_constraint(self):52        """Tests that referred column in a foreign key (ex. id) is found"""53        main_table = 'test_reverse_ref'54        reference_table = 'test_rr_foreign'55        db.start_transaction()56        self._create_foreign_tables(main_table, reference_table)57        db.execute_deferred_sql()58        inverse = db._lookup_reverse_constraint(reference_table, 'id')59        (cname, rev_table, rev_column) = inverse[0]60        self.assertEquals(main_table, rev_table)61        self.assertEquals('foreign_id', rev_column)62        db.delete_table(main_table)63        db.delete_table(reference_table)64    def test_delete_fk_column(self):65        main_table = 'test_drop_foreign'66        ref_table = 'test_df_ref'67        self._create_foreign_tables(main_table, ref_table)68        db.execute_deferred_sql()69        constraints = db._find_foreign_constraints(main_table, 'foreign_id')70        self.assertEquals(len(constraints), 1)71        db.delete_column(main_table, 'foreign_id')72        constraints = db._find_foreign_constraints(main_table, 'foreign_id')73        self.assertEquals(len(constraints), 0)74        db.delete_table(main_table)75        db.delete_table(ref_table)76    def test_rename_fk_column(self):77        main_table = 'test_rename_foreign'78        ref_table = 'test_rf_ref'79        self._create_foreign_tables(main_table, ref_table)80        db.execute_deferred_sql()81        constraints = db._find_foreign_constraints(main_table, 'foreign_id')82        self.assertEquals(len(constraints), 1)83        db.rename_column(main_table, 'foreign_id', 'reference_id')84        db.execute_deferred_sql()  #Create constraints85        constraints = db._find_foreign_constraints(main_table, 'reference_id')86        self.assertEquals(len(constraints), 1)87        db.delete_table(main_table)88        db.delete_table(ref_table)89    def test_rename_fk_inbound(self):90        """91        Tests that the column referred to by an external column can be renamed.92        Edge case, but also useful as stepping stone to renaming tables.93        """94        main_table = 'test_rename_fk_inbound'95        ref_table = 'test_rfi_ref'96        self._create_foreign_tables(main_table, ref_table)97        db.execute_deferred_sql()98        constraints = db._lookup_reverse_constraint(ref_table, 'id')99        self.assertEquals(len(constraints), 1)100        db.rename_column(ref_table, 'id', 'rfi_id')101        db.execute_deferred_sql()  #Create constraints102        constraints = db._lookup_reverse_constraint(ref_table, 'rfi_id')103        self.assertEquals(len(constraints), 1)104        cname = db._find_foreign_constraints(main_table, 'foreign_id')[0]105        (rtable, rcolumn) = db._lookup_constraint_references(main_table, cname)106        self.assertEquals(rcolumn, 'rfi_id')107        db.delete_table(main_table)108        db.delete_table(ref_table)109    def test_rename_constrained_table(self):110        """Renames a table with a foreign key column (towards another table)"""111        main_table = 'test_rn_table'112        ref_table = 'test_rt_ref'113        renamed_table = 'test_renamed_table'114        self._create_foreign_tables(main_table, ref_table)115        db.execute_deferred_sql()116        constraints = db._find_foreign_constraints(main_table, 'foreign_id')117        self.assertEquals(len(constraints), 1)118        db.rename_table(main_table, renamed_table)119        db.execute_deferred_sql()  #Create constraints120        constraints = db._find_foreign_constraints(renamed_table, 'foreign_id')121        self.assertEquals(len(constraints), 1)122        (rtable, rcolumn) = db._lookup_constraint_references(123                renamed_table, constraints[0])124        self.assertEquals(rcolumn, 'id')125        db.delete_table(renamed_table)126        db.delete_table(ref_table)127    def test_renamed_referenced_table(self):128        """Rename a table referred to in a foreign key"""129        main_table = 'test_rn_refd_table'130        ref_table = 'test_rrt_ref'131        renamed_table = 'test_renamed_ref'132        self._create_foreign_tables(main_table, ref_table)133        db.execute_deferred_sql()134        constraints = db._lookup_reverse_constraint(ref_table)135        self.assertEquals(len(constraints), 1)136        db.rename_table(ref_table, renamed_table)137        db.execute_deferred_sql()  #Create constraints138        constraints = db._find_foreign_constraints(main_table, 'foreign_id')139        self.assertEquals(len(constraints), 1)140        (rtable, rcolumn) = db._lookup_constraint_references(141                main_table, constraints[0])142        self.assertEquals(renamed_table, rtable)143        db.delete_table(main_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.