You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
stacosys/tests/test_db.py

118 lines
3.9 KiB
Python

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import time
import pytest
from stacosys.db import dao, database
from stacosys.model.comment import Comment
@pytest.fixture
def setup_db():
database.configure("sqlite:memory://db.sqlite")
def equals_comment(comment: Comment, other):
return (
comment.id == other.id
and comment.author_gravatar == other.author_gravatar
and comment.author_name == other.author_name
and comment.author_site == other.author_site
and comment.content == other.content
and comment.created == other.created
and comment.notified == other.notified
and comment.published == other.published
)
def test_find_comment_by_id(setup_db):
assert dao.find_comment_by_id(1) is None
c1 = dao.create_comment("/post1", "Yax", "", "", "Comment 1")
assert c1.id is not None
find_c1 = dao.find_comment_by_id(c1.id)
assert find_c1
assert equals_comment(c1, find_c1)
c1.id = find_c1.id
dao.delete_comment(c1)
assert dao.find_comment_by_id(c1.id) is None
def test_dao_published(setup_db):
assert 0 == dao.count_published_comments("")
c1 = dao.create_comment("/post1", "Yax", "", "", "Comment 1")
assert 0 == dao.count_published_comments("")
assert 1 == len(dao.find_not_published_comments())
dao.publish_comment(c1)
assert 1 == dao.count_published_comments("")
c2 = dao.create_comment("/post2", "Yax", "", "", "Comment 2")
dao.publish_comment(c2)
assert 2 == dao.count_published_comments("")
c3 = dao.create_comment("/post2", "Yax", "", "", "Comment 3")
dao.publish_comment(c3)
assert 0 == len(dao.find_not_published_comments())
# count published
assert 1 == dao.count_published_comments("/post1")
assert 2 == dao.count_published_comments("/post2")
# find published
assert 0 == len(dao.find_published_comments_by_url("/"))
assert 1 == len(dao.find_published_comments_by_url("/post1"))
assert 2 == len(dao.find_published_comments_by_url("/post2"))
def test_dao_notified(setup_db):
assert 0 == len(dao.find_not_notified_comments())
c1 = dao.create_comment("/post1", "Yax", "", "", "Comment 1")
assert 1 == len(dao.find_not_notified_comments())
c2 = dao.create_comment("/post2", "Yax", "", "", "Comment 2")
assert 2 == len(dao.find_not_notified_comments())
dao.notify_comment(c1)
dao.notify_comment(c2)
assert 0 == len(dao.find_not_notified_comments())
c3 = dao.create_comment("/post2", "Yax", "", "", "Comment 3")
assert 1 == len(dao.find_not_notified_comments())
dao.notify_comment(c3)
assert 0 == len(dao.find_not_notified_comments())
def create_comment(url, author_name, content):
return dao.create_comment(url, author_name, "", "", content)
def test_find_recent_published_comments(setup_db):
comments = [create_comment("/post", "Adam", "Comment 1"), create_comment("/post", "Arf", "Comment 2"),
create_comment("/post", "Arwin", "Comment 3"), create_comment("/post", "Bill", "Comment 4"),
create_comment("/post", "Bo", "Comment 5"), create_comment("/post", "Charles", "Comment 6"),
create_comment("/post", "Dan", "Comment 7"), create_comment("/post", "Dwayne", "Comment 8"),
create_comment("/post", "Erl", "Comment 9"), create_comment("/post", "Jay", "Comment 10"),
create_comment("/post", "Kenny", "Comment 11"), create_comment("/post", "Lord", "Comment 12")]
rows = dao.find_recent_published_comments()
assert len(rows) == 0
# publish every second
for comment in comments:
dao.publish_comment(comment)
time.sleep(1)
rows = dao.find_recent_published_comments()
assert len(rows) == 10
authors = [row.author_name for row in rows]
assert authors == [
"Lord",
"Kenny",
"Jay",
"Erl",
"Dwayne",
"Dan",
"Charles",
"Bo",
"Bill",
"Arwin",
]