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

Copied to Clipboard!

How to use test_delete_table method in localstack

Best Python code snippet using localstack_python

test_gcp_bigquery.py

Source:test_gcp_bigquery.py

1from .config import *2import time3from datetime import date4from bibtutils.gcp import bigquery, storage5def test_bq_delete():6    bigquery.delete_table(7        TEST_PROJECT, TEST_DATASET, "test_delete_table", not_found_ok=True8    )9    test_schema = [10        {"name": "test_name", "mode": "REQUIRED", "type": "STRING"},11        {"name": "upload_date", "mode": "NULLABLE", "type": "DATE"},12    ]13    bigquery.create_table(14        TEST_PROJECT,15        TEST_DATASET,16        "test_delete_table",17        schema_json=test_schema,18        time_partitioning_interval="HOUR",19    )20    bigquery.delete_table(TEST_PROJECT, TEST_DATASET, "test_delete_table")21def test_bigquery():22    test_timestamp = int(time.time())23    test_data = [24        {25            'test_name': 'test_storage_nldjson',26            'test_desc': 'Test data for NLD JSON GCS interactions.',27            'timestamp': test_timestamp28        },29        {30            'test_name': 'test_storage_nldjson_r2',31            'test_desc': 'Test data for NLD JSON GCS interactions. Row 2.',32            'timestamp': test_timestamp33        }34    ]35    test_blob = f'test-bq-upload-{test_timestamp}.nldjson'36    storage.write_gcs_nldjson(37        bucket_name=TEST_BUCKET,38        blob_name=test_blob,39        json_data=test_data,40        add_date=True,41    )42    bigquery.upload_gcs_json(43        bucket_name=TEST_BUCKET,44        blob_name=test_blob,45        bq_project=TEST_PROJECT,46        dataset=TEST_DATASET,47        table=TEST_TABLE,48        append=True,49        ignore_unknown=False,50        autodetect_schema=True51    )52    rcvd_data = bigquery.query(53        f'select * from `{TEST_PROJECT}.{TEST_DATASET}.{TEST_TABLE}` '54        f'where timestamp={test_timestamp}'55    )56    for i in range(len(test_data)):57        test_data[i]['upload_date'] = date.today()58    assert rcvd_data == test_data59def test_bigquery_create_table():60    test_schema = [61        {62            'name': 'test_name',63            'mode': 'REQUIRED',64            'type': 'STRING'65        },66        {67            'name': 'upload_date',68            'mode': 'NULLABLE',69            'type': 'DATE'70        }71    ]72    bigquery.create_table(73        TEST_PROJECT, TEST_DATASET, 'test_create_table',74        schema_json=test_schema,75        time_partitioning_interval='HOUR',76    )77    # cleanup78    from google.cloud import bigquery as bq79    bq.Client().delete_table(f'{TEST_PROJECT}.{TEST_DATASET}.test_create_table', not_found_ok=False)80def test_bigquery_create_and_upload():81    test_timestamp = int(time.time())82    test_data = [83        {84            'test_name': 'test_create_upload',85            'test_desc': 'Test data for NLD JSON GCS table create/upload.',86            'timestamp': test_timestamp87        },88        {89            'test_name': 'test_create_upload_r2',90            'test_desc': 'Test data for NLD JSON GCS table create/upload. Row 2.',91            'timestamp': test_timestamp92        }93    ]94    test_schema = [95        {96            'name': 'test_name',97            'mode': 'REQUIRED',98            'type': 'STRING'99        },100        {101            'name': 'test_desc',102            'mode': 'NULLABLE',103            'type': 'STRING'104        },105        {106            'name': 'timestamp',107            'mode': 'NULLABLE',108            'type': 'INTEGER'109        },110        {111            'name': 'upload_date',112            'mode': 'NULLABLE',113            'type': 'DATE'114        }115    ]116    test_blob = f'test-bq-create-and-upload-{test_timestamp}.nldjson'117    storage.write_gcs_nldjson(118        bucket_name=TEST_BUCKET,119        blob_name=test_blob,120        json_data=test_data,121        add_date=True,122    )123    # prep124    from google.cloud import bigquery as bq125    client = bq.Client()126    client.delete_table(f'{TEST_PROJECT}.{TEST_DATASET}.test_create_upload_table', not_found_ok=True)127    # test128    bigquery.create_and_upload(129        bucket_name=TEST_BUCKET,130        blob_name=test_blob,131        bq_project=TEST_PROJECT,132        dataset=TEST_DATASET,133        table='test_create_upload_table',134        append=False,135        ignore_unknown=False,136        autodetect_schema=True,137        schema_json=test_schema,138        time_partitioning_interval='DAY',139        time_partitioning_field='upload_date',140        already_created_ok=False141    )142    # cleanup...

test_order_db.py

Source:test_order_db.py

...35    my_cursor.execute(query)36    res = my_cursor.fetchall() #storing the output of fetchall() in the res37    for row in res: #printing all the records in res38        print(row)39def test_delete_table():40    query = "SET FOREIGN_KEY_CHECKS=0; DROP TABLE tbl_order"41    my_cursor.execute(query)42    print("table successfully deleted")43    44test_append()45test_read()...

test_db_operations.py

Source:test_db_operations.py

...3from src import db_operations4class DynamoDBTestCase(unittest.TestCase):5    6    @mock_dynamodb7    def test_delete_table(self):8        result = db_operations.delete_table()9        self.assertEqual(result["message"], "deleted table")10    @mock_dynamodb11    def test_create_table(self):12        result = db_operations.create_visitor_table()13        self.assertEqual(result.table_name, 'Visitors')14    @mock_dynamodb15    def test_delete_table(self):16        result = db_operations.delete_table()17        self.assertEqual(result["message"], "deleted table")18    @mock_dynamodb19    def test_initialize_count(self):20        result = db_operations.initialize_count("count")21        self.assertEqual(result["ResponseMetadata"]["HTTPStatusCode"], 200)22    @mock_dynamodb23    def test_get_count(self):24        result = db_operations.get_count("count",)25        self.assertEqual(result["visitor_total"], 0)26    @mock_dynamodb27    def test_update_count(self):28        result = db_operations.update_count("count",)29        self.assertEqual(result['Attributes']["visitor_total"], 1)...

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.