How to use sql_runner method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_index.py

Source:test_index.py

1import pytest2from sql_runner import SQLRunner3from sql_runner_selects import SQLRunnerSelects4from sql_selects import *5sql_runner_select = SQLRunnerSelects()6table_select = sql_runner_select.execute_create_file()7table_select = sql_runner_select.execute_seed_file()8def test_create_table():9    sql_runner = SQLRunner()10    table = sql_runner.execute_create_file()11    planets = table.execute("SELECT name FROM sqlite_master WHERE type='table';").fetchone()[0]12    results = table.execute("PRAGMA table_info('%s')" % planets).fetchall()13    columns = []14    for el in results:15        cleaned_col = (el[1], el[2])16        columns.append(cleaned_col)17    # id18    assert columns[0][0] == 'id', 'id not set to Primary Key'19    assert columns[0][1].upper() == 'INTEGER', 'id not set to Primary Key'20    # name21    assert columns[1][0] == 'name', 'name not set to TEXT'22    assert columns[1][1].upper() == 'TEXT', 'name not set to TEXT'23    # color24    assert columns[2][0] == 'color', 'color not set to TEXT'25    assert columns[2][1].upper() == 'TEXT', 'color not set to TEXT'26    # num_of_moons27    assert columns[3][0] == 'num_of_moons', 'num_of_moons not set to INTEGER'28    assert columns[3][1].upper() == 'INTEGER', 'num_of_moons not set to INTEGER'29    # mass30    assert columns[4][0] == 'mass', 'mass not set to REAL'31    assert columns[4][1].upper() == 'REAL', 'mass not set to REAL'32def test_alter_table():33    sql_runner = SQLRunner()34    altered_table = sql_runner.execute_alter_file()35    planets = altered_table.execute("SELECT name FROM sqlite_master WHERE type='table';").fetchone()[0]36    results = altered_table.execute("PRAGMA table_info('%s')" % planets).fetchall()37    assert results[-1][1] == 'rings', 'rings not set to BOOLEAN'38    assert results[-1][2] == 'BOOLEAN', 'rings not set to BOOLEAN'39def test_insert_into():40    sql_runner = SQLRunner()41    table = sql_runner.execute_alter_file()42    table_values = sql_runner.execute_insert_file()43    test = table_values.execute("SELECT * FROM planets").fetchall()44    assert len(test) == 945def test_update_jupiter():46    sql_runner = SQLRunner()47    table = sql_runner.execute_alter_file()48    table_values = sql_runner.execute_insert_file()49    update = sql_runner.execute_update_file()50    result = 6851    assert table_values.execute("SELECT num_of_moons FROM planets WHERE name = 'Jupiter';").fetchone()[0] == result52def test_delete_from():53    sql_runner = SQLRunner()54    table = sql_runner.execute_alter_file()55    table_values = sql_runner.execute_insert_file()56    deletion = sql_runner.execute_delete_file()57    test_delete = deletion.execute("SELECT * FROM planets").fetchall()58    assert len(test_delete) == 8, "Delete Pluto!"59def test_select_all_columns_and_rows():60    result = [(1, 'Mercury', 'gray', 0, 0.55, 0.0), (2, 'Venus', 'yellow', 0, 0.82, 0.0), (3, 'Earth', 'blue', 1, 1.0, 0.0), (4, 'Mars', 'red', 2, 0.11, 0.0), (5, 'Jupiter', 'orange', 67, 317.9, 0.0), (6, 'Saturn', 'hazel', 62, 95.19, 1.0), (7, 'Uranus', 'light blue', 27, 14.54, 1.0), (8, 'Neptune', 'dark blue', 14, 17.15, 1.0)]61    assert table_select.execute(select_all_columns_and_rows()).fetchall() == result62def test_select_name_and_color_of_all_planets():63    result = [('Mercury', 'gray'), ('Venus', 'yellow'), ('Earth', 'blue'), ('Mars', 'red'), ('Jupiter', 'orange'), ('Saturn', 'hazel'), ('Uranus', 'light blue'), ('Neptune', 'dark blue')]64    assert table_select.execute(select_name_and_color_of_all_planets()).fetchall() == result65def test_select_all_planets_with_mass_greater_than_one():66    result = [(5, 'Jupiter', 'orange', 67, 317.9, 0.0), (6, 'Saturn', 'hazel', 62, 95.19, 1.0), (7, 'Uranus', 'light blue', 27, 14.54, 1.0), (8, 'Neptune', 'dark blue', 14, 17.15, 1.0)]67    assert table_select.execute(select_all_planets_with_mass_greater_than_one()).fetchall() == result68def test_select_name_and_mass_of_planets_with_mass_less_than_equal_to_one():69    result = [('Mercury', 0.55), ('Venus', 0.82), ('Earth', 1.0), ('Mars', 0.11)]70    assert table_select.execute(select_name_and_mass_of_planets_with_mass_less_than_equal_to_one()).fetchall() == result71def test_select_name_and_color_of_planets_with_more_than_10_moons():72    result = [('Jupiter', 'orange'), ('Saturn', 'hazel'), ('Uranus', 'light blue'), ('Neptune', 'dark blue')]73    assert table_select.execute(select_name_and_color_of_planets_with_more_than_10_moons()).fetchall() == result74def test_select_all_planets_with_moons_and_mass_less_than_one():75    result = [(4, 'Mars', 'red', 2, 0.11, 0.0)]76    assert table_select.execute(select_all_planets_with_moons_and_mass_less_than_one()).fetchall() == result77def test_select_name_and_color_of_all_blue_planets():78    result = [('Earth', 'blue'), ('Uranus', 'light blue'), ('Neptune', 'dark blue')]...

test_sql_utils.py

Source:test_sql_utils.py

...3import bgt_setup4from fme.sql_utils import SQLRunner5log = logging.getLogger(__name__)6@pytest.fixture7def sql_runner():8    # 5401 / localhost9    runner = SQLRunner(10        host=bgt_setup.DB_FME_HOST, port=bgt_setup.DB_FME_PORT,11        dbname=bgt_setup.DB_FME_DBNAME, user=bgt_setup.DB_FME_USER)12    runner.run_sql("DROP TABLE IF EXISTS public.sql_utils;")13    return runner14@pytest.mark.skip(reason='integration test')15def test_sql_runner_run_sql(sql_runner):16    dml_tekst = "CREATE TABLE public.sql_utils(field1 CHARACTER VARYING(8), field2 CHARACTER VARYING(38), field3 DATE);"17    sql_runner.run_sql(dml_tekst)18    res = sql_runner.run_sql('SELECT * FROM public.sql_utils')19    assert len(res) == 020    res = sql_runner.run_sql(21        "INSERT INTO public.sql_utils (field1, field2, field3)"...

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.