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

How to use assert_rows_equal method in Testify

Best Python code snippet using Testify_python

test_assertions.py

Source:test_assertions.py

...98        try:99            assert_not_reached()100        except AssertionError:101            pass102    def test_assert_rows_equal(self):103        row1 = dict(a=1, b=2)104        row2 = dict(b=2, a=1)105        assert_rows_equal([row1, row2], [row2, row1])106        row1 = [1, 2]107        row2 = [3, 4]108        assert_rows_equal([row1, row2], [row2, row1])109    def test_assert_length(self):110        assert_length('abc', 3)111    def test_assert_is(self):112        x = 3113        y = x114        assert_is(x, y)115        x = 300116        assert_is(x, 300)117        assert_is(None, None)118        from testy.aliases import eq119        assert_is(eq, assert_equal)120    def test_assert_is_not(self):121        assert_is_not(assert_is, assert_is_not)122        assert_is_not('abc', list('abc'))123        l = [1, 2, 3]124        assert_is_not(l, l[:])125    def test_assert_all_match_regex(self):126        values = [127            'abc',128            '123 abc def',129        ]130        pattern = re.compile(r'\w+')131        assert_all_match_regex(pattern, values)132    def test_assert_match_regex(self):133        pattern = re.compile(r'\w+')134        assert_match_regex(pattern, 'abc 123')135    def test_assert_any_match_regex(self):136        values = [137            '"$',138            'abc',139            '@#~',140        ]141        pattern = re.compile(r'\w+')142        assert_any_match_regex(pattern, values)143    def test_assert_all_not_match_regex(self):144        values = [145            '"$',146            '@#~',147        ]148        pattern = re.compile(r'\w+')149        assert_all_not_match_regex(pattern, values)150    def test_assert_sets_equal(self):151        s1 = set(['a', 'b', 'c', 1, 2, 3])152        s2 = set([1, 'a', 3, 'b', 'c', 2])153        assert_sets_equal(s1, s2)154    def test_assert_dicts_equal(self):155        d1 = dict(a=3, b=True, c=None)156        d2 = dict(b=True, c=None, a=3)157        assert_dicts_equal(d1, d2)158        d1 = dict(a=3, b=True, c=None, d=4)159        d2 = dict(b=True, c=None, a=3)160        assert_dicts_equal(d1, d2, ignore_keys=['d'])161    def test_assert_dict_subset(self):162        d1 = dict(b=True)163        d2 = dict(a=3, b=True, c=None)164        assert_dict_subset(d1, d2)165    def test_assert_subset(self):166        s1 = set([3, 'b', 'c', 2])167        s2 = set(['a', 'b', 'c', 1, 2, 3])168        assert_subset(s1, s2)169    def test_assert_list_prefix(self):170        l1 = [1, 2, 3]171        l2 = [1, 2, 3, 'a', 'b', 'c']172        assert_list_prefix(l1, l2)173    def test_assert_sorted_equal(self):174        s1 = set(['a', 'b', 'c'])175        s2 = set(['b', 'c', 'a'])176        assert_sorted_equal(s1, s2)177    def test_assert_isinstance(self):178        class A(object):179            pass180        assert_isinstance(A(), A)181        assert_isinstance(dict(a=1), dict)182    def test_assert_datetimes_equal(self):183        # times are compared to the millisecond, so this ought to pass184        t0 = datetime.now()185        t1 = datetime.now()186        assert_datetimes_equal(t0, t1)187        t0 = datetime(1970, 1, 1)188        t1 = datetime(1970, 1, 1)189        assert_datetimes_equal(t0, t1)190    def test_assert_exactly_one(self):191        assert_exactly_one(None, False, None, None)192        assert_exactly_one(None, True, None, None)193class NegativeAssertionsTestCase(unittest.TestCase):194    """Test all assertions with the expectation of them all failing."""195    def test_assert_raises(self):196        class MyException(Exception):197            pass198        with assert_raises(AssertionError):199            with assert_raises(TypeError):200                raise MyException()201        with assert_raises(AssertionError):202            with assert_raises(Exception):203                pass204    def test_assert_raises_and_contains(self):205        def no_fail():206            return207        def fail():208            raise ValueError("choose one of the correct values")209        with assert_raises(AssertionError):210            assert_raises_and_contains(ValueError, "two of", fail)211        with assert_raises(AssertionError):212            assert_raises_and_contains(Exception, "anything", no_fail)213    def test_assert_equal(self):214        with assert_raises(AssertionError):215            assert_equal(1, 2)216    def test_assert_almost_equal(self):217        with assert_raises(AssertionError):218            assert_almost_equal(1, 1.01, 2)219    def test_assert_within_tolerance(self):220        with assert_raises(AssertionError):221            assert_within_tolerance(5, 5.1, 0.01)222    def test_assert_not_equal(self):223        with assert_raises(AssertionError):224            assert_not_equal(1, 1)225    def test_assert_lt(self):226        with assert_raises(AssertionError):227            assert_lt(3, 2)228    def test_assert_lte(self):229        with assert_raises(AssertionError):230            assert_lte(10, 1)231    def test_assert_gt(self):232        with assert_raises(AssertionError):233            assert_gt(1, 4)234    def test_assert_gte(self):235        with assert_raises(AssertionError):236            assert_gte(3, 5)237    def test_assert_in_range(self):238        with assert_raises(AssertionError):239            assert_in_range(1, 2, 4)240    def test_assert_between(self):241        with assert_raises(AssertionError):242            assert_between(1, 3, 2)243    def test_assert_in(self):244        with assert_raises(AssertionError):245            assert_in('a', [1, 2, 3])246    def test_assert_not_in(self):247        with assert_raises(AssertionError):248            assert_not_in(1, [1, 2, 3])249    def test_assert_all_in(self):250        with assert_raises(AssertionError):251            assert_all_in([1, 2], [1, 3])252    def test_assert_starts_with(self):253        with assert_raises(AssertionError):254            assert_starts_with('abc123', 'bc')255    def test_assert_not_reached(self):256        # The only way to test this assertion negatively is to not reach it :)257        pass258    def test_assert_rows_equal(self):259        with assert_raises(AssertionError):260            row1 = dict(a=1, b=2)261            row2 = dict(b=3, a=1)262            row3 = dict(b=1, a=1)263            assert_rows_equal([row1, row2], [row2, row3])264    def test_assert_length(self):265        with assert_raises(AssertionError):266            assert_length('abc', 4)267    def test_assert_is(self):268        with assert_raises(AssertionError):269            assert_is(True, False)270    def test_assert_is_not(self):271        with assert_raises(AssertionError):272            assert_is_not(True, True)273    def test_assert_all_match_regex(self):274        with assert_raises(AssertionError):275            values = [276                '$%`',277                '123 abc def',...

tools.py

Source:tools.py

...47    def _write_and_diff_helper(data: List[dict], update_num: int):48        get_target_writer(engine, True)({str(table.name): ([], data)})49        result = get_data_for_comparison(engine)50        _, expected_data = get_expected_data(update_num)51        assert_rows_equal(expected_data, result)52    _write_and_diff_helper(TEST_DATA_INITIAL, FIRST_UPDATE)53    _write_and_diff_helper(TEST_DATA_APPEND_SINGLE_ROW, SECOND_UPDATE)54    _write_and_diff_helper(TEST_DATA_APPEND_MULTIPLE_ROWS, THIRD_UPDATE)55    _write_and_diff_helper(TEST_DATA_UPDATE_SINGLE_ROW, FOURTH_UPDATE)56def validate_drop_primary_keys(engine: Engine, table: Table):57    """58    Verify that dropping a primary key from using drop_primary_keys leaves a relational database in the required state.59    :param engine:60    :param table:61    :return:62    """63    with engine.connect() as conn:64        conn.execute(table.insert(), TEST_DATA_APPEND_MULTIPLE_ROWS)65    pks_to_drop = [{'first_name': 'Stefanos', 'last_name': 'Tsitsipas'}]66    drop_primary_keys(engine, table, pks_to_drop)67    result = get_data_for_comparison(engine)68    assert_rows_equal(TEST_DATA_APPEND_MULTIPLE_ROWS_WITH_DELETE, result)69def validate_dolt_as_target(db_engine: Engine,70                            db_table: Table,71                            get_db_source_reader,72                            get_db_target_writer,73                            get_db_table_reader,74                            dssc,75                            dolt_table,76                            datetime_strict: bool = True):77    """78    Validates syncing from a relational database, so far MySQL and Postgres, to Dolt. Work by making a series of writes79    to the relational database (running in a Docker container provided by a fixture), executing a sync, and then80    validating the HEAD of main of the Dolt repo has the expected values. It also validates that the Dolt history is81    correct after every write. Finally validates that deletes flow through to Dolt.82    :param db_engine:83    :param db_table:84    :param get_db_source_reader:85    :param get_db_target_writer:86    :param get_db_table_reader:87    :param dssc:88    :param dolt_table:89    :param datetime_strict:90    :return:91    """92    def sync_to_dolt_helper():93        source_reader = get_db_source_reader(db_engine, get_db_table_reader())94        target_writer = get_dolt_target_writer(dssc, commit=True)95        sync_to_dolt(source_reader, target_writer, {str(db_table.name): str(dolt_table.name)})96    def assertion_helper(commit: str, expected_diff: List[dict]):97        """98        Validates that both the HEAD of the current branch of the Dolt repo match MySQL, and that the difIfs created by99        the write match what is expected.100        """101        _, dolt_data = get_dolt_table_reader(commit)(str(dolt_table.name), dssc)102        db_table_metadata = get_table_metadata(db_engine, str(db_table.name))103        db_data = get_db_table_reader()(db_engine, db_table_metadata)104        assert_rows_equal(list(dolt_data), db_data, datetime_strict=datetime_strict)105        _, dolt_diff_data = get_dolt_table_reader_diffs(commit)(str(dolt_table.name), dssc)106        assert_rows_equal(expected_diff, list(dolt_diff_data), datetime_strict=datetime_strict)107    update_sequence = [108        TEST_DATA_INITIAL,109        TEST_DATA_APPEND_MULTIPLE_ROWS,110        TEST_DATA_APPEND_SINGLE_ROW,111        TEST_DATA_UPDATE_SINGLE_ROW112    ]113    for update_data in update_sequence:114        get_db_target_writer(db_engine)({str(db_table.name): ([], update_data)})115        sync_to_dolt_helper()116        latest_commit = list(dssc.log().keys())[0]117        assertion_helper(latest_commit, update_data)118    with db_engine.connect() as conn:119        conn.execute(db_table.delete().where(db_table.c.first_name == 'Novak'))120    sync_to_dolt_helper()121    latest_commit = list(dssc.log().keys())[0]122    _, dolt_data = get_dolt_table_reader(latest_commit)(str(dolt_table.name), dssc)123    db_data = get_db_table_reader()(db_engine, db_table)124    assert_rows_equal(list(dolt_data), db_data, datetime_strict=datetime_strict)125    dropped_pks, _ = get_dolt_table_reader_diffs(latest_commit)(str(dolt_table.name), dssc)126    assert dropped_pks == [{'first_name': 'Novak', 'last_name': 'Djokovic'}]127def validate_dolt_as_source(db_conn, db_table, get_db_target_writer, dolt_repo, dolt_table):128    """129    Verifies that given a Dolt repository that has a a series of updates applied to it (defined in the fixture130    create_dolt_test_data_commits) that after syncing at each of the commits, the Dolt repository and the target131    relational database, so far MySQL or Postgres server instance contain the same data. Tests creates, updates, and132    deletes.133    :param db_conn:134    :param db_table:135    :param get_db_target_writer:136    :param dolt_repo:137    :param dolt_table:138    :return:139    """140    target_writer = get_db_target_writer(db_conn, True)141    table_mapping = {str(dolt_table.name): str(db_table.name)}142    commits = list(dolt_repo.log().keys())143    commits_to_check = [commits[0], commits[1], commits[2], commits[3], commits[4]]144    commits_to_check.reverse()145    for i, commit in enumerate(commits_to_check):146        logger.info('Syncing from Dolt at commit {}, {} of {}'.format(commit, i, len(commits_to_check)))147        table_reader = get_dolt_table_reader_diffs(commit)148        sync_from_dolt(get_dolt_source_reader(dolt_repo, table_reader), target_writer, table_mapping)149        db_data = get_data_for_comparison(db_conn)150        _, dolt_data = get_dolt_table_reader(commit)(str(dolt_table.name), dolt_repo)...

test_dolt.py

Source:test_dolt.py

...51        logger.info('comparison for commit/update_num {}/{}'.format(commit, update_num))52        dropped_pks, dolt_data = build_table_reader(commit)(str(table.name), dssc)53        expected_dropped_pks, expected_data = get_expected(update_num)54        assert expected_dropped_pks == dropped_pks55        assert_rows_equal(expected_data, list(dolt_data))56def test_get_target_writer(db_with_table):57    """58    When writing to Dolt from a relational database we want to replicate the state of the database at each commit, since59    the database itself stores no history, we must read the entire dataset each time, and delete appropriate PKs60    :param db_with_table:61    :return:62    """63    dssc, dolt_table = db_with_table64    update_sequence = [65        TEST_DATA_INITIAL,66        TEST_DATA_INITIAL + TEST_DATA_APPEND_MULTIPLE_ROWS,67        TEST_DATA_APPEND_MULTIPLE_ROWS + TEST_DATA_APPEND_SINGLE_ROW,68        TEST_DATA_APPEND_MULTIPLE_ROWS + TEST_DATA_UPDATE_SINGLE_ROW69    ]70    for i, update in enumerate(update_sequence):71        logger.info('Making {} of {} updates and validating'.format(i, len(update_sequence)))72        get_target_writer(dssc, commit=True)({str(dolt_table.name): update})73        result = _dolt_table_read_helper(dssc, str(dolt_table.name))74        assert_rows_equal(update, result)75def _dolt_table_read_helper(dssc: DoltSQLServerContext, table_name: str):76    table = get_table_metadata(dssc.engine, table_name)77    with dssc.engine.connect() as conn:78        result = conn.execute(table.select())...

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.