Codebase list anosql / lintian-fixes/main tests / test_sqlite3.py
lintian-fixes/main

Tree @lintian-fixes/main (Download .tar.gz)

test_sqlite3.py @lintian-fixes/main — raw · history · blame 

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
import os

import anosql
import pytest


def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d


@pytest.fixture()
def queries():
    dir_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "blogdb", "sql")
    return anosql.from_path(dir_path, "sqlite3")


def test_record_query(sqlite3_conn, queries):
    sqlite3_conn.row_factory = dict_factory
    actual = queries.users.get_all(sqlite3_conn)

    assert len(actual) == 3
    assert actual[0] == {
        "userid": 1,
        "username": "bobsmith",
        "firstname": "Bob",
        "lastname": "Smith",
    }


def test_parameterized_query(sqlite3_conn, queries):
    actual = queries.blogs.get_user_blogs(sqlite3_conn, userid=1)
    expected = [("How to make a pie.", "2018-11-23"), ("What I did Today", "2017-07-28")]
    assert actual == expected


def test_parameterized_record_query(sqlite3_conn, queries):
    sqlite3_conn.row_factory = dict_factory
    actual = queries.blogs.sqlite_get_blogs_published_after(sqlite3_conn, published="2018-01-01")

    expected = [
        {"title": "How to make a pie.", "username": "bobsmith", "published": "2018-11-23 00:00"},
        {"title": "Testing", "username": "janedoe", "published": "2018-01-01 00:00"},
    ]

    assert actual == expected


def test_select_cursor_context_manager(sqlite3_conn, queries):
    with queries.blogs.get_user_blogs_cursor(sqlite3_conn, userid=1) as cursor:
        actual = cursor.fetchall()
        expected = [("How to make a pie.", "2018-11-23"), ("What I did Today", "2017-07-28")]
        assert actual == expected


def test_insert_returning(sqlite3_conn, queries):
    with sqlite3_conn:
        blogid = queries.blogs.publish_blog(
            sqlite3_conn,
            userid=2,
            title="My first blog",
            content="Hello, World!",
            published="2018-12-04",
        )
    cur = sqlite3_conn.cursor()
    cur.execute(
        """\
        select title
          from blogs
         where blogid = ?;
    """,
        (blogid,),
    )
    actual = cur.fetchone()
    cur.close()
    expected = ("My first blog",)

    assert actual == expected


def test_delete(sqlite3_conn, queries):
    # Removing the "janedoe" blog titled "Testing"
    actual = queries.blogs.remove_blog(sqlite3_conn, blogid=2)
    assert actual is None

    janes_blogs = queries.blogs.get_user_blogs(sqlite3_conn, userid=3)
    assert len(janes_blogs) == 0


def test_insert_many(sqlite3_conn, queries):
    blogs = [
        (2, "Blog Part 1", "content - 1", "2018-12-04"),
        (2, "Blog Part 2", "content - 2", "2018-12-05"),
        (2, "Blog Part 3", "content - 3", "2018-12-06"),
    ]

    with sqlite3_conn:
        actual = queries.blogs.sqlite_bulk_publish(sqlite3_conn, blogs)
        assert actual is None

        johns_blogs = queries.blogs.get_user_blogs(sqlite3_conn, userid=2)
        assert johns_blogs == [
            ("Blog Part 3", "2018-12-06"),
            ("Blog Part 2", "2018-12-05"),
            ("Blog Part 1", "2018-12-04"),
        ]