102 lines
3 KiB
Python
102 lines
3 KiB
Python
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",
|
|
)
|
|
print(blogid, type(blogid))
|
|
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"),
|
|
]
|