124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
#!/usr/bin/env 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",
|
|
]
|