test_sqlite.py

import pytest
from cbutil.postprocessing.sqlite_helper import (select_stmt,
                                                 from_stmt,
                                                 where_stmt,
                                                 join_stmt,
                                                 table_name_query,
                                                 query_builder)

from cbutil.postprocessing.sqlite import (get_all_table_names,
                                          tables2dict,
                                          query_join,
                                          iterate_all_tables)

from cbutil.postprocessing.sqlite import sqlite_context


def test_select():
    assert select_stmt().strip() == "SELECT *"
    assert select_stmt("test").strip() == "SELECT test"
    with pytest.raises(ValueError):
        select_stmt("")
    with pytest.raises(ValueError):
        select_stmt("     ")


def test_from():
    assert from_stmt("table").strip() == "FROM table"
    with pytest.raises(ValueError):
        from_stmt("")
    with pytest.raises(ValueError):
        from_stmt("    ")


def test_where():
    lhs = "lhs"
    rhs = "rhs"
    assert where_stmt(lhs, rhs).strip() == f"WHERE {lhs}={rhs}"
    with pytest.raises(ValueError):
        where_stmt("", lhs)
    with pytest.raises(ValueError):
        where_stmt("    ", lhs)
    with pytest.raises(ValueError):
        where_stmt(rhs, "")
    with pytest.raises(ValueError):
        where_stmt(rhs, "    ")
    with pytest.raises(ValueError):
        where_stmt(rhs, lhs, "")
    with pytest.raises(ValueError):
        where_stmt(rhs, lhs, "    ")


def test_join():
    lhs = "lhs"
    rhs = "rhs"
    key = "key"
    assert join_stmt(lhs, rhs, key).strip() == f"{lhs} inner join {rhs} on {lhs}.{key} = {rhs}.{key}"


def test_table_name():
    assert table_name_query() == "SELECT name FROM sqlite_master WHERE type='table'"


def test_builder():
    assert query_builder(from_table="table") == "SELECT * FROM table"
    assert query_builder(select="row", from_table="table") == "SELECT row FROM table"


def test_all_table_names():
    with sqlite_context("tests/benchmark.sqlite") as connection:
        names = get_all_table_names(connection)
        assert sorted(["runs", "timingPool"]) == sorted(names)


def test_join_query():
    with sqlite_context("tests/benchmark.sqlite") as connection:
        dicts = list(tables2dict(query_join(connection, "runs", "timingPool", "runId")))
        assert len(dicts) == 9


def test_iterate_query():
    dicts = list(tables2dict(iterate_all_tables("tests/cpu_benchmark.sqlite3")))
    assert len(dicts) == 150