summaryrefslogtreecommitdiff
path: root/server
diff options
context:
space:
mode:
authorrr- <rr-@sakuya.pl>2016-05-07 21:42:03 +0200
committerrr- <rr-@sakuya.pl>2016-05-07 21:50:37 +0200
commit58964bcdc9cfeccd72a017fdbd2f4ffc8ba7b6f3 (patch)
tree1996c877f0ad62e56605fe259a6cd6294bcb59e7 /server
parent9b591c3f1ba36f8aabee57d7ce46457acbb20794 (diff)
server/posts: add post listing
Diffstat (limited to 'server')
-rw-r--r--server/szurubooru/api/post_api.py11
-rw-r--r--server/szurubooru/db/post.py39
-rw-r--r--server/szurubooru/func/util.py9
-rw-r--r--server/szurubooru/search/__init__.py1
-rw-r--r--server/szurubooru/search/base_search_config.py34
-rw-r--r--server/szurubooru/search/comment_search_config.py2
-rw-r--r--server/szurubooru/search/post_search_config.py175
-rw-r--r--server/szurubooru/search/search_executor.py31
-rw-r--r--server/szurubooru/tests/api/test_post_retrieving.py53
-rw-r--r--server/szurubooru/tests/search/test_post_search_config.py578
10 files changed, 898 insertions, 35 deletions
diff --git a/server/szurubooru/api/post_api.py b/server/szurubooru/api/post_api.py
index ed049f3..e848555 100644
--- a/server/szurubooru/api/post_api.py
+++ b/server/szurubooru/api/post_api.py
@@ -1,8 +1,19 @@
import datetime
+from szurubooru import search
from szurubooru.api.base_api import BaseApi
from szurubooru.func import auth, tags, posts, snapshots, favorites, scores
class PostListApi(BaseApi):
+ def __init__(self):
+ super().__init__()
+ self._search_executor = search.SearchExecutor(search.PostSearchConfig())
+
+ def get(self, ctx):
+ auth.verify_privilege(ctx.user, 'posts:list')
+ self._search_executor.config.user = ctx.user
+ return self._search_executor.execute_and_serialize(
+ ctx, lambda post: posts.serialize_post(post, ctx.user))
+
def post(self, ctx):
auth.verify_privilege(ctx.user, 'posts:create')
content = ctx.get_file('content', required=True)
diff --git a/server/szurubooru/db/post.py b/server/szurubooru/db/post.py
index 7affa73..081e97f 100644
--- a/server/szurubooru/db/post.py
+++ b/server/szurubooru/db/post.py
@@ -2,6 +2,7 @@ from sqlalchemy import Column, Integer, DateTime, String, Text, PickleType, Fore
from sqlalchemy.orm import relationship, column_property, object_session
from sqlalchemy.sql.expression import func, select
from szurubooru.db.base import Base
+from szurubooru.db.comment import Comment
class PostFeature(Base):
__tablename__ = 'post_feature'
@@ -117,6 +118,8 @@ class Post(Base):
.where(PostTag.post_id == post_id) \
.correlate_except(PostTag))
+ canvas_area = column_property(canvas_width * canvas_height)
+
@property
def is_featured(self):
featured_post = object_session(self) \
@@ -125,12 +128,10 @@ class Post(Base):
.first()
return featured_post and featured_post.post_id == self.post_id
- @property
- def score(self):
- return object_session(self) \
- .query(func.sum(PostScore.score)) \
- .filter(PostScore.post_id == self.post_id) \
- .one()[0] or 0
+ score = column_property(
+ select([func.coalesce(func.sum(PostScore.score), 0)]) \
+ .where(PostScore.post_id == post_id) \
+ .correlate_except(PostScore))
favorite_count = column_property(
select([func.count(PostFavorite.post_id)]) \
@@ -152,10 +153,22 @@ class Post(Base):
.where(PostFeature.post_id == post_id) \
.correlate_except(PostFeature))
- # TODO: wire these
- #comment_count = Column('auto_comment_count', Integer, nullable=False, default=0)
- #note_count = Column('auto_note_count', Integer, nullable=False, default=0)
- #last_comment_edit_time = Column(
- # 'auto_comment_creation_time', Integer, nullable=False, default=0)
- #last_comment_creation_time = Column(
- # 'auto_comment_edit_time', Integer, nullable=False, default=0)
+ comment_count = column_property(
+ select([func.count(Comment.post_id)]) \
+ .where(Comment.post_id == post_id) \
+ .correlate_except(Comment))
+
+ last_comment_creation_time = column_property(
+ select([func.max(Comment.creation_time)]) \
+ .where(Comment.post_id == post_id) \
+ .correlate_except(Comment))
+
+ last_comment_edit_time = column_property(
+ select([func.max(Comment.last_edit_time)]) \
+ .where(Comment.post_id == post_id) \
+ .correlate_except(Comment))
+
+ note_count = column_property(
+ select([func.count(PostNote.post_id)]) \
+ .where(PostNote.post_id == post_id) \
+ .correlate_except(PostNote))
diff --git a/server/szurubooru/func/util.py b/server/szurubooru/func/util.py
index 2fcddd8..a80485d 100644
--- a/server/szurubooru/func/util.py
+++ b/server/szurubooru/func/util.py
@@ -4,6 +4,15 @@ import re
from sqlalchemy.inspection import inspect
from szurubooru.errors import ValidationError
+def unalias_dict(input_dict):
+ output_dict = {}
+ for key_list, value in input_dict.items():
+ if isinstance(key_list, str):
+ key_list = [key_list]
+ for key in key_list:
+ output_dict[key] = value
+ return output_dict
+
def get_md5(source):
if not isinstance(source, bytes):
source = source.encode('utf-8')
diff --git a/server/szurubooru/search/__init__.py b/server/szurubooru/search/__init__.py
index 14db2bc..d16e987 100644
--- a/server/szurubooru/search/__init__.py
+++ b/server/szurubooru/search/__init__.py
@@ -5,3 +5,4 @@ from szurubooru.search.user_search_config import UserSearchConfig
from szurubooru.search.snapshot_search_config import SnapshotSearchConfig
from szurubooru.search.tag_search_config import TagSearchConfig
from szurubooru.search.comment_search_config import CommentSearchConfig
+from szurubooru.search.post_search_config import PostSearchConfig
diff --git a/server/szurubooru/search/base_search_config.py b/server/szurubooru/search/base_search_config.py
index 530b85a..3822199 100644
--- a/server/szurubooru/search/base_search_config.py
+++ b/server/szurubooru/search/base_search_config.py
@@ -1,8 +1,12 @@
import sqlalchemy
import szurubooru.errors
+from szurubooru import db
from szurubooru.func import util
from szurubooru.search import criteria
+def wildcard_transformer(value):
+ return value.replace('*', '%')
+
class BaseSearchConfig(object):
SORT_DESC = -1
SORT_ASC = 1
@@ -50,17 +54,16 @@ class BaseSearchConfig(object):
BaseSearchConfig._apply_num_criterion_to_column(column, criterion))
@staticmethod
- def _apply_str_criterion_to_column(column, criterion):
+ def _apply_str_criterion_to_column(column, criterion, transformer):
'''
Decorate SQLAlchemy filter on given column using supplied criterion.
- Parse potential wildcards inside the criterion.
'''
if isinstance(criterion, criteria.PlainSearchCriterion):
- expr = column.like(criterion.value.replace('*', '%'))
+ expr = column.like(transformer(criterion.value))
elif isinstance(criterion, criteria.ArraySearchCriterion):
expr = sqlalchemy.sql.false()
for value in criterion.values:
- expr = expr | column.like(value.replace('*', '%'))
+ expr = expr | column.like(transformer(value))
elif isinstance(criterion, criteria.RangedSearchCriterion):
raise szurubooru.errors.SearchError(
'Composite token %r is invalid in this context.' % (criterion,))
@@ -71,9 +74,10 @@ class BaseSearchConfig(object):
return expr
@staticmethod
- def _create_str_filter(column):
+ def _create_str_filter(column, transformer=wildcard_transformer):
return lambda query, criterion: query.filter(
- BaseSearchConfig._apply_str_criterion_to_column(column, criterion))
+ BaseSearchConfig._apply_str_criterion_to_column(
+ column, criterion, transformer))
@staticmethod
def _apply_date_criterion_to_column(column, criterion):
@@ -111,3 +115,21 @@ class BaseSearchConfig(object):
def _create_date_filter(column):
return lambda query, criterion: query.filter(
BaseSearchConfig._apply_date_criterion_to_column(column, criterion))
+
+ @staticmethod
+ def _create_subquery_filter(
+ left_id_column,
+ right_id_column,
+ filter_column,
+ filter_factory,
+ subquery_decorator=None):
+ filter_func = filter_factory(filter_column)
+ def func(query, criterion):
+ subquery = db.session.query(right_id_column.label('foreign_id'))
+ if subquery_decorator:
+ subquery = subquery_decorator(subquery)
+ subquery = subquery.options(sqlalchemy.orm.lazyload('*'))
+ subquery = filter_func(subquery, criterion)
+ subquery = subquery.subquery('t')
+ return query.filter(left_id_column == subquery.c.foreign_id)
+ return func
diff --git a/server/szurubooru/search/comment_search_config.py b/server/szurubooru/search/comment_search_config.py
index c01979d..52fa779 100644
--- a/server/szurubooru/search/comment_search_config.py
+++ b/server/szurubooru/search/comment_search_config.py
@@ -19,6 +19,7 @@ class CommentSearchConfig(BaseSearchConfig):
'id': self._create_num_filter(db.Comment.comment_id),
'post': self._create_num_filter(db.Comment.post_id),
'user': self._create_str_filter(db.User.name),
+ 'author': self._create_str_filter(db.User.name),
'text': self._create_str_filter(db.Comment.text),
'creation-date': self._create_date_filter(db.Comment.creation_time),
'creation-time': self._create_date_filter(db.Comment.creation_time),
@@ -33,6 +34,7 @@ class CommentSearchConfig(BaseSearchConfig):
return {
'random': (func.random(), None),
'user': (db.User.name, self.SORT_ASC),
+ 'author': (db.User.name, self.SORT_ASC),
'post': (db.Comment.post_id, self.SORT_DESC),
'creation-date': (db.Comment.creation_time, self.SORT_DESC),
'creation-time': (db.Comment.creation_time, self.SORT_DESC),
diff --git a/server/szurubooru/search/post_search_config.py b/server/szurubooru/search/post_search_config.py
new file mode 100644
index 0000000..79dc7ba
--- /dev/null
+++ b/server/szurubooru/search/post_search_config.py
@@ -0,0 +1,175 @@
+from sqlalchemy.sql.expression import func
+from szurubooru import db, errors
+from szurubooru.func import util
+from szurubooru.search.base_search_config import BaseSearchConfig
+
+def _type_transformer(value):
+ available_types = {
+ 'image': db.Post.TYPE_IMAGE,
+ 'animation': db.Post.TYPE_ANIMATION,
+ 'animated': db.Post.TYPE_ANIMATION,
+ 'anim': db.Post.TYPE_ANIMATION,
+ 'gif': db.Post.TYPE_ANIMATION,
+ 'video': db.Post.TYPE_VIDEO,
+ 'webm': db.Post.TYPE_VIDEO,
+ 'flash': db.Post.TYPE_FLASH,
+ 'swf': db.Post.TYPE_FLASH,
+ }
+ try:
+ return available_types[value.lower()]
+ except KeyError:
+ raise errors.SearchError('Invalid type: %r. Available types: %r.' % (
+ value, available_types))
+
+class PostSearchConfig(BaseSearchConfig):
+ def create_query(self):
+ return db.session.query(db.Post)
+
+ def finalize_query(self, query):
+ return query.order_by(db.Post.creation_time.desc())
+
+ @property
+ def anonymous_filter(self):
+ return self._create_subquery_filter(
+ db.Post.post_id,
+ db.PostTag.post_id,
+ db.TagName.name,
+ self._create_str_filter,
+ lambda subquery: subquery.join(db.Tag).join(db.TagName))
+
+ @property
+ def named_filters(self):
+ return util.unalias_dict({
+ 'id': self._create_num_filter(db.Post.post_id),
+ 'tag': self._create_subquery_filter(
+ db.Post.post_id,
+ db.PostTag.post_id,
+ db.TagName.name,
+ self._create_str_filter,
+ lambda subquery: subquery.join(db.Tag).join(db.TagName)),
+ 'score': self._create_num_filter(db.Post.score),
+ ('uploader', 'upload', 'submit'):
+ self._create_subquery_filter(
+ db.Post.user_id,
+ db.User.user_id,
+ db.User.name,
+ self._create_str_filter),
+ 'comment': self._create_subquery_filter(
+ db.Post.post_id,
+ db.Comment.post_id,
+ db.User.name,
+ self._create_str_filter,
+ lambda subquery: subquery.join(db.User)),
+ 'fav': self._create_subquery_filter(
+ db.Post.post_id,
+ db.PostFavorite.post_id,
+ db.User.name,
+ self._create_str_filter,
+ lambda subquery: subquery.join(db.User)),
+ 'tag-count': self._create_num_filter(db.Post.tag_count),
+ 'comment-count': self._create_num_filter(db.Post.comment_count),
+ 'fav-count': self._create_num_filter(db.Post.favorite_count),
+ 'note-count': self._create_num_filter(db.Post.note_count),
+ 'feature-count': self._create_num_filter(db.Post.feature_count),
+ 'type': self._create_str_filter(db.Post.type, _type_transformer),
+ 'file-size': self._create_num_filter(db.Post.file_size),
+ ('image-width', 'width'):
+ self._create_num_filter(db.Post.canvas_width),
+ ('image-height', 'height'):
+ self._create_num_filter(db.Post.canvas_height),
+ ('image-area', 'area'):
+ self._create_num_filter(db.Post.canvas_area),
+ ('creation-date', 'creation-time', 'date', 'time'):
+ self._create_date_filter(db.Post.creation_time),
+ ('last-edit-date', 'last-edit-time', 'edit-date', 'edit-time'):
+ self._create_date_filter(db.Post.last_edit_time),
+ ('comment-date', 'comment-time'):
+ self._create_date_filter(db.Post.last_comment_edit_time),
+ ('fav-date', 'fav-time'):
+ self._create_date_filter(db.Post.last_favorite_time),
+ ('feature-date', 'feature-time'):
+ self._create_date_filter(db.Post.last_feature_time),
+ })
+
+ @property
+ def sort_columns(self):
+ return util.unalias_dict({
+ 'random': (func.random(), None),
+ 'id': (db.Post.post_id, self.SORT_DESC),
+ 'score': (db.Post.score, self.SORT_DESC),
+ 'tag-count': (db.Post.tag_count, self.SORT_DESC),
+ 'comment-count': (db.Post.comment_count, self.SORT_DESC),
+ 'fav-count': (db.Post.favorite_count, self.SORT_DESC),
+ 'note-count': (db.Post.note_count, self.SORT_DESC),
+ 'feature-count': (db.Post.feature_count, self.SORT_DESC),
+ 'file-size': (db.Post.file_size, self.SORT_DESC),
+ ('image-width', 'width'): (db.Post.canvas_width, self.SORT_DESC),
+ ('image-height', 'height'): (db.Post.canvas_height, self.SORT_DESC),
+ ('image-area', 'area'): (db.Post.canvas_area, self.SORT_DESC),
+ ('creation-date', 'creation-time', 'date', 'time'):
+ (db.Post.creation_time, self.SORT_DESC),
+ ('last-edit-date', 'last-edit-time', 'edit-date', 'edit-time'):
+ (db.Post.last_edit_time, self.SORT_DESC),
+ ('comment-date', 'comment-time'):
+ (db.Post.last_comment_edit_time, self.SORT_DESC),
+ ('fav-date', 'fav-time'):
+ (db.Post.last_favorite_time, self.SORT_DESC),
+ ('feature-date', 'feature-time'):
+ (db.Post.last_feature_time, self.SORT_DESC),
+ })
+
+ @property
+ def special_filters(self):
+ return {
+ 'liked': self.own_liked_filter,
+ 'disliked': self.own_disliked_filter,
+ 'fav': self.own_fav_filter,
+ 'tumbleweed': self.tumbleweed_filter,
+ }
+
+ def own_liked_filter(self, query, negated):
+ assert self.user
+ if self.user.rank == 'anonymous':
+ raise errors.SearchError('Must be logged in to use this feature.')
+ expr = db.Post.post_id.in_(
+ db.session \
+ .query(db.PostScore.post_id) \
+ .filter(db.PostScore.user_id == self.user.user_id) \
+ .filter(db.PostScore.score == 1))
+ if negated:
+ expr = ~expr
+ return query.filter(expr)
+
+ def own_disliked_filter(self, query, negated):
+ assert self.user
+ if self.user.rank == 'anonymous':
+ raise errors.SearchError('Must be logged in to use this feature.')
+ expr = db.Post.post_id.in_(
+ db.session \
+ .query(db.PostScore.post_id) \
+ .filter(db.PostScore.user_id == self.user.user_id) \
+ .filter(db.PostScore.score == -1))
+ if negated:
+ expr = ~expr
+ return query.filter(expr)
+
+ def own_fav_filter(self, query, negated):
+ assert self.user
+ if self.user.rank == 'anonymous':
+ raise errors.SearchError('Must be logged in to use this feature.')
+ expr = db.Post.post_id.in_(
+ db.session \
+ .query(db.PostFavorite.post_id) \
+ .filter(db.PostFavorite.user_id == self.user.user_id))
+ if negated:
+ expr = ~expr
+ return query.filter(expr)
+
+ def tumbleweed_filter(self, query, negated):
+ expr = \
+ (db.Post.comment_count == 0) \
+ & (db.Post.favorite_count == 0) \
+ & (db.Post.score == 0)
+ if negated:
+ expr = ~expr
+ return query.filter(expr)
diff --git a/server/szurubooru/search/search_executor.py b/server/szurubooru/search/search_executor.py
index 7a3ac8a..bade7a4 100644
--- a/server/szurubooru/search/search_executor.py
+++ b/server/szurubooru/search/search_executor.py
@@ -10,7 +10,7 @@ class SearchExecutor(object):
'''
def __init__(self, search_config):
- self._search_config = search_config
+ self.config = search_config
def execute(self, query_text, page, page_size):
'''
@@ -43,7 +43,7 @@ class SearchExecutor(object):
def _prepare(self, query_text):
''' Parse input and return SQLAlchemy query. '''
- query = self._search_config.create_query() \
+ query = self.config.create_query() \
.options(sqlalchemy.orm.lazyload('*'))
for token in re.split(r'\s+', (query_text or '').lower()):
if not token:
@@ -60,7 +60,7 @@ class SearchExecutor(object):
query = self._handle_anonymous(
query, self._create_criterion(token, negated))
- query = self._search_config.finalize_query(query)
+ query = self.config.finalize_query(query)
return query
def _handle_key_value(self, query, key, value, negated):
@@ -72,10 +72,10 @@ class SearchExecutor(object):
return self._handle_named(query, key, value, negated)
def _handle_anonymous(self, query, criterion):
- if not self._search_config.anonymous_filter:
+ if not self.config.anonymous_filter:
raise errors.SearchError(
'Anonymous tokens are not valid in this context.')
- return self._search_config.anonymous_filter(query, criterion)
+ return self.config.anonymous_filter(query, criterion)
def _handle_named(self, query, key, value, negated):
if key.endswith('-min'):
@@ -85,19 +85,18 @@ class SearchExecutor(object):
key = key[:-4]
value = '..' + value
criterion = self._create_criterion(value, negated)
- if key in self._search_config.named_filters:
- return self._search_config.named_filters[key](query, criterion)
+ if key in self.config.named_filters:
+ return self.config.named_filters[key](query, criterion)
raise errors.SearchError(
'Unknown named token: %r. Available named tokens: %r.' % (
- key, list(self._search_config.named_filters.keys())))
+ key, list(self.config.named_filters.keys())))
def _handle_special(self, query, value, negated):
- if value in self._search_config.special_filters:
- return self._search_config.special_filters[value](
- query, value, negated)
+ if value in self.config.special_filters:
+ return self.config.special_filters[value](query, negated)
raise errors.SearchError(
'Unknown special token: %r. Available special tokens: %r.' % (
- value, list(self._search_config.special_filters.keys())))
+ value, list(self.config.special_filters.keys())))
def _handle_sort(self, query, value, negated):
if value.count(',') == 0:
@@ -108,14 +107,14 @@ class SearchExecutor(object):
raise errors.SearchError('Too many commas in sort style token.')
try:
- column, default_sort = self._search_config.sort_columns[value]
+ column, default_sort = self.config.sort_columns[value]
except KeyError:
raise errors.SearchError(
'Unknown sort style: %r. Available sort styles: %r.' % (
- value, list(self._search_config.sort_columns.keys())))
+ value, list(self.config.sort_columns.keys())))
- sort_asc = self._search_config.SORT_ASC
- sort_desc = self._search_config.SORT_DESC
+ sort_asc = self.config.SORT_ASC
+ sort_desc = self.config.SORT_DESC
try:
sort_map = {
diff --git a/server/szurubooru/tests/api/test_post_retrieving.py b/server/szurubooru/tests/api/test_post_retrieving.py
index 369fa79..b32e52d 100644
--- a/server/szurubooru/tests/api/test_post_retrieving.py
+++ b/server/szurubooru/tests/api/test_post_retrieving.py
@@ -21,9 +21,62 @@ def test_ctx(
ret.context_factory = context_factory
ret.user_factory = user_factory
ret.post_factory = post_factory
+ ret.list_api = api.PostListApi()
ret.detail_api = api.PostDetailApi()
return ret
+def test_retrieving_multiple(test_ctx):
+ post1 = test_ctx.post_factory(id=1)
+ post2 = test_ctx.post_factory(id=2)
+ db.session.add_all([post1, post2])
+ result = test_ctx.list_api.get(
+ test_ctx.context_factory(
+ input={'query': '', 'page': 1},
+ user=test_ctx.user_factory(rank='regular_user')))
+ assert result['query'] == ''
+ assert result['page'] == 1
+ assert result['pageSize'] == 100
+ assert result['total'] == 2
+ assert [t['id'] for t in result['results']] == [1, 2]
+
+def test_using_special_tokens(
+ test_ctx, config_injector):
+ auth_user = test_ctx.user_factory(rank='regular_user')
+ post1 = test_ctx.post_factory(id=1)
+ post2 = test_ctx.post_factory(id=2)
+ post1.favorited_by = [db.PostFavorite(
+ user=auth_user, time=datetime.datetime.now())]
+ db.session.add_all([post1, post2, auth_user])
+ db.session.flush()
+ result = test_ctx.list_api.get(
+ test_ctx.context_factory(
+ input={'query': 'special:fav', 'page': 1},
+ user=auth_user))
+ assert result['query'] == 'special:fav'
+ assert result['page'] == 1
+ assert result['pageSize'] == 100
+ assert result['total'] == 1
+ assert [t['id'] for t in result['results']] == [1]
+
+def test_trying_to_use_special_tokens_without_logging_in(
+ test_ctx, config_injector):
+ config_injector({
+ 'privileges': {'posts:list': 'anonymous'},
+ 'ranks': ['anonymous'],
+ })
+ with pytest.raises(errors.SearchError):
+ test_ctx.list_api.get(
+ test_ctx.context_factory(
+ input={'query': 'special:fav', 'page': 1},
+ user=test_ctx.user_factory(rank='anonymous')))
+
+def test_trying_to_retrieve_multiple_without_privileges(test_ctx):
+ with pytest.raises(errors.AuthError):
+ test_ctx.list_api.get(
+ test_ctx.context_factory(
+ input={'query': '', 'page': 1},
+ user=test_ctx.user_factory(rank='anonymous')))
+
def test_retrieving_single(test_ctx):
db.session.add(test_ctx.post_factory(id=1))
result = test_ctx.detail_api.get(
diff --git a/server/szurubooru/tests/search/test_post_search_config.py b/server/szurubooru/tests/search/test_post_search_config.py
new file mode 100644
index 0000000..2a9e8bb
--- /dev/null
+++ b/server/szurubooru/tests/search/test_post_search_config.py
@@ -0,0 +1,578 @@
+import datetime
+import pytest
+from szurubooru import db, errors, search
+
+@pytest.fixture
+def fav_factory(user_factory):
+ def factory(post, user=None):
+ return db.PostFavorite(
+ post=post, user=user or user_factory(), time=datetime.datetime.now())
+ return factory
+
+@pytest.fixture
+def score_factory(user_factory):
+ def factory(post, user=None, score=1):
+ return db.PostScore(
+ post=post,
+ user=user or user_factory(),
+ time=datetime.datetime.now(),
+ score=score)
+ return factory
+
+@pytest.fixture
+def note_factory():
+ def factory(post=None):
+ if post:
+ return db.PostNote(polygon='...', text='...', post=post)
+ return db.PostNote(polygon='...', text='...')
+ return factory
+
+@pytest.fixture
+def feature_factory(user_factory):
+ def factory(post=None):
+ if post:
+ return db.PostFeature(
+ time=datetime.datetime.now(), user=user_factory(), post=post)
+ return db.PostFeature(time=datetime.datetime.now(), user=user_factory())
+ return factory
+
+@pytest.fixture
+def executor(user_factory):
+ return search.SearchExecutor(search.PostSearchConfig())
+
+@pytest.fixture
+def auth_executor(executor, user_factory):
+ def wrapper():
+ auth_user = user_factory()
+ db.session.add(auth_user)
+ db.session.flush()
+ executor.config.user = auth_user
+ return auth_user
+ return wrapper
+
+@pytest.fixture
+def verify_unpaged(executor):
+ def verify(input, expected_post_ids, test_order=False):
+ actual_count, actual_posts = executor.execute(
+ input, page=1, page_size=100)
+ actual_post_ids = list([p.post_id for p in actual_posts])
+ print(actual_post_ids, expected_post_ids)
+ assert actual_count == len(expected_post_ids)
+ if not test_order:
+ actual_post_ids = sorted(actual_post_ids)
+ expected_post_ids = sorted(expected_post_ids)
+ assert actual_post_ids == expected_post_ids
+ return verify
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('id:1', [1]),
+ ('id:3', [3]),
+ ('id:1,3', [1, 3]),
+])
+def test_filter_by_id(verify_unpaged, post_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('tag:t1', [1]),
+ ('tag:t2', [2]),
+ ('tag:t1,t2', [1, 2]),
+ ('tag:t4a', [4]),
+ ('tag:t4b', [4]),
+])
+def test_filter_by_tag(
+ verify_unpaged, post_factory, tag_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post4 = post_factory(id=4)
+ post1.tags=[tag_factory(names=['t1'])]
+ post2.tags=[tag_factory(names=['t2'])]
+ post3.tags=[tag_factory(names=['t3'])]
+ post4.tags=[tag_factory(names=['t4a', 't4b'])]
+ db.session.add_all([post1, post2, post3, post4])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('score:1', [1]),
+ ('score:3', [3]),
+ ('score:1,3', [1, 3]),
+])
+def test_filter_by_score(
+ verify_unpaged, post_factory, user_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ for post in [post1, post2, post3]:
+ db.session.add(
+ db.PostScore(
+ score=post.post_id,
+ time=datetime.datetime.now(),
+ post=post,
+ user=user_factory()))
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('uploader:u1', [1]),
+ ('uploader:u3', [3]),
+ ('uploader:u1,u3', [1, 3]),
+ ('upload:u1', [1]),
+ ('upload:u3', [3]),
+ ('upload:u1,u3', [1, 3]),
+ ('submit:u1', [1]),
+ ('submit:u3', [3]),
+ ('submit:u1,u3', [1, 3]),
+])
+def test_filter_by_uploader(
+ verify_unpaged, post_factory, user_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.user = user_factory(name='u1')
+ post2.user = user_factory(name='u2')
+ post3.user = user_factory(name='u3')
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('comment:u1', [1]),
+ ('comment:u3', [3]),
+ ('comment:u1,u3', [1, 3]),
+])
+def test_filter_by_commenter(
+ verify_unpaged,
+ post_factory,
+ user_factory,
+ comment_factory,
+ input,
+ expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([
+ comment_factory(post=post1, user=user_factory(name='u1')),
+ comment_factory(post=post2, user=user_factory(name='u2')),
+ comment_factory(post=post3, user=user_factory(name='u3')),
+ post1, post2, post3,
+ ])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('fav:u1', [1]),
+ ('fav:u3', [3]),
+ ('fav:u1,u3', [1, 3]),
+])
+def test_filter_by_favorite(
+ verify_unpaged,
+ post_factory,
+ user_factory,
+ fav_factory,
+ input,
+ expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([
+ fav_factory(post=post1, user=user_factory(name='u1')),
+ fav_factory(post=post2, user=user_factory(name='u2')),
+ fav_factory(post=post3, user=user_factory(name='u3')),
+ post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('tag-count:1', [1]),
+ ('tag-count:3', [3]),
+ ('tag-count:1,3', [1, 3]),
+])
+def test_filter_by_tag_count(
+ verify_unpaged, post_factory, tag_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.tags=[tag_factory()]
+ post2.tags=[tag_factory(), tag_factory()]
+ post3.tags=[tag_factory(), tag_factory(), tag_factory()]
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('comment-count:1', [1]),
+ ('comment-count:3', [3]),
+ ('comment-count:1,3', [1, 3]),
+])
+def test_filter_by_comment_count(
+ verify_unpaged, post_factory, comment_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([
+ comment_factory(post=post1),
+ comment_factory(post=post2),
+ comment_factory(post=post2),
+ comment_factory(post=post3),
+ comment_factory(post=post3),
+ comment_factory(post=post3),
+ post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('fav-count:1', [1]),
+ ('fav-count:3', [3]),
+ ('fav-count:1,3', [1, 3]),
+])
+def test_filter_by_favorite_count(
+ verify_unpaged, post_factory, fav_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([
+ fav_factory(post=post1),
+ fav_factory(post=post2),
+ fav_factory(post=post2),
+ fav_factory(post=post3),
+ fav_factory(post=post3),
+ fav_factory(post=post3),
+ post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('note-count:1', [1]),
+ ('note-count:3', [3]),
+ ('note-count:1,3', [1, 3]),
+])
+def test_filter_by_note_count(
+ verify_unpaged, post_factory, note_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.notes=[note_factory()]
+ post2.notes=[note_factory(), note_factory()]
+ post3.notes=[note_factory(), note_factory(), note_factory()]
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('feature-count:1', [1]),
+ ('feature-count:3', [3]),
+ ('feature-count:1,3', [1, 3]),
+])
+def test_filter_by_feature_count(
+ verify_unpaged, post_factory, feature_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.features=[feature_factory()]
+ post2.features=[feature_factory(), feature_factory()]
+ post3.features=[feature_factory(), feature_factory(), feature_factory()]
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('type:image', [1]),
+ ('type:anim', [2]),
+ ('type:animation', [2]),
+ ('type:gif', [2]),
+ ('type:video', [3]),
+ ('type:webm', [3]),
+ ('type:flash', [4]),
+ ('type:swf', [4]),
+])
+def test_filter_by_type(verify_unpaged, post_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post4 = post_factory(id=4)
+ post1.type = db.Post.TYPE_IMAGE
+ post2.type = db.Post.TYPE_ANIMATION
+ post3.type = db.Post.TYPE_VIDEO
+ post4.type = db.Post.TYPE_FLASH
+ db.session.add_all([post1, post2, post3, post4])
+ verify_unpaged(input, expected_post_ids)
+
+def test_filter_by_invalid_type(executor):
+ with pytest.raises(errors.SearchError):
+ actual_count, actual_posts = executor.execute(
+ 'type:invalid', page=1, page_size=100)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('file-size:100', [1]),
+ ('file-size:102', [3]),
+ ('file-size:100,102', [1, 3]),
+])
+def test_filter_by_file_size(
+ verify_unpaged, post_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.file_size = 100
+ post2.file_size = 101
+ post3.file_size = 102
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('image-width:100', [1]),
+ ('image-width:102', [3]),
+ ('image-width:100,102', [1, 3]),
+ ('image-height:200', [1]),
+ ('image-height:202', [3]),
+ ('image-height:200,202', [1, 3]),
+ ('image-area:20000', [1]),
+ ('image-area:20604', [3]),
+ ('image-area:20000,20604', [1, 3]),
+])
+def test_filter_by_image_size(
+ verify_unpaged, post_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.canvas_width = 100
+ post2.canvas_width = 101
+ post3.canvas_width = 102
+ post1.canvas_height = 200
+ post2.canvas_height = 201
+ post3.canvas_height = 202
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('creation-date:2014', [1]),
+ ('creation-date:2016', [3]),
+ ('creation-date:2014,2016', [1, 3]),
+ ('creation-time:2014', [1]),
+ ('creation-time:2016', [3]),
+ ('creation-time:2014,2016', [1, 3]),
+ ('date:2014', [1]),
+ ('date:2016', [3]),
+ ('date:2014,2016', [1, 3]),
+ ('time:2014', [1]),
+ ('time:2016', [3]),
+ ('time:2014,2016', [1, 3]),
+])
+def test_filter_by_creation_time(
+ verify_unpaged, post_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.creation_time = datetime.datetime(2014, 1, 1)
+ post2.creation_time = datetime.datetime(2015, 1, 1)
+ post3.creation_time = datetime.datetime(2016, 1, 1)
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('last-edit-date:2014', [1]),
+ ('last-edit-date:2016', [3]),
+ ('last-edit-date:2014,2016', [1, 3]),
+ ('last-edit-time:2014', [1]),
+ ('last-edit-time:2016', [3]),
+ ('last-edit-time:2014,2016', [1, 3]),
+ ('edit-date:2014', [1]),
+ ('edit-date:2016', [3]),
+ ('edit-date:2014,2016', [1, 3]),
+ ('edit-time:2014', [1]),
+ ('edit-time:2016', [3]),
+ ('edit-time:2014,2016', [1, 3]),
+])
+def test_filter_by_last_edit_time(
+ verify_unpaged, post_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post1.last_edit_time = datetime.datetime(2014, 1, 1)
+ post2.last_edit_time = datetime.datetime(2015, 1, 1)
+ post3.last_edit_time = datetime.datetime(2016, 1, 1)
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('comment-date:2014', [1]),
+ ('comment-date:2016', [3]),
+ ('comment-date:2014,2016', [1, 3]),
+ ('comment-time:2014', [1]),
+ ('comment-time:2016', [3]),
+ ('comment-time:2014,2016', [1, 3]),
+])
+def test_filter_by_comment_date(
+ verify_unpaged, post_factory, comment_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ comment1 = comment_factory(post=post1)
+ comment2 = comment_factory(post=post2)
+ comment3 = comment_factory(post=post3)
+ comment1.last_edit_time = datetime.datetime(2014, 1, 1)
+ comment2.last_edit_time = datetime.datetime(2015, 1, 1)
+ comment3.last_edit_time = datetime.datetime(2016, 1, 1)
+ db.session.add_all([post1, post2, post3, comment1, comment2, comment3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('fav-date:2014', [1]),
+ ('fav-date:2016', [3]),
+ ('fav-date:2014,2016', [1, 3]),
+ ('fav-time:2014', [1]),
+ ('fav-time:2016', [3]),
+ ('fav-time:2014,2016', [1, 3]),
+])
+def test_filter_by_fav_date(
+ verify_unpaged, post_factory, fav_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ fav1 = fav_factory(post=post1)
+ fav2 = fav_factory(post=post2)
+ fav3 = fav_factory(post=post3)
+ fav1.time = datetime.datetime(2014, 1, 1)
+ fav2.time = datetime.datetime(2015, 1, 1)
+ fav3.time = datetime.datetime(2016, 1, 1)
+ db.session.add_all([post1, post2, post3, fav1, fav2, fav3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('feature-date:2014', [1]),
+ ('feature-date:2016', [3]),
+ ('feature-date:2014,2016', [1, 3]),
+ ('feature-time:2014', [1]),
+ ('feature-time:2016', [3]),
+ ('feature-time:2014,2016', [1, 3]),
+])
+def test_filter_by_feature_date(
+ verify_unpaged, post_factory, feature_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ feature1 = feature_factory(post=post1)
+ feature2 = feature_factory(post=post2)
+ feature3 = feature_factory(post=post3)
+ feature1.time = datetime.datetime(2014, 1, 1)
+ feature2.time = datetime.datetime(2015, 1, 1)
+ feature3.time = datetime.datetime(2016, 1, 1)
+ db.session.add_all([post1, post2, post3, feature1, feature2, feature3])
+ verify_unpaged(input, expected_post_ids)
+
+@pytest.mark.parametrize('input', [
+ 'sort:random',
+ 'sort:id',
+ 'sort:score',
+ 'sort:tag-count',
+ 'sort:comment-count',
+ 'sort:fav-count',
+ 'sort:note-count',
+ 'sort:feature-count',
+ 'sort:file-size',
+ 'sort:image-width',
+ 'sort:width',
+ 'sort:image-height',
+ 'sort:height',
+ 'sort:image-area',
+ 'sort:area',
+ 'sort:creation-date',
+ 'sort:creation-time',
+ 'sort:date',
+ 'sort:time',
+ 'sort:last-edit-date',
+ 'sort:last-edit-time',
+ 'sort:edit-date',
+ 'sort:edit-time',
+ 'sort:comment-date',
+ 'sort:comment-time',
+ 'sort:fav-date',
+ 'sort:fav-time',
+ 'sort:feature-date',
+ 'sort:feature-time',
+])
+def test_sort_tokens(verify_unpaged, post_factory, input):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([post1, post2, post3])
+ verify_unpaged(input, [1, 2, 3])
+
+@pytest.mark.parametrize('input,expected_post_ids', [
+ ('', [1, 2, 3, 4]),
+ ('t1', [1]),
+ ('t2', [2]),
+ ('t1,t2', [1, 2]),
+ ('t4a', [4]),
+ ('t4b', [4]),
+])
+def test_anonymous(
+ verify_unpaged, post_factory, tag_factory, input, expected_post_ids):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post4 = post_factory(id=4)
+ post1.tags=[tag_factory(names=['t1'])]
+ post2.tags=[tag_factory(names=['t2'])]
+ post3.tags=[tag_factory(names=['t3'])]
+ post4.tags=[tag_factory(names=['t4a', 't4b'])]
+ db.session.add_all([post1, post2, post3, post4])
+ verify_unpaged(input, expected_post_ids)
+
+def test_own_liked(
+ auth_executor, post_factory, score_factory, user_factory, verify_unpaged):
+ auth_user = auth_executor()
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([
+ score_factory(post=post1, user=auth_user, score=1),
+ score_factory(post=post2, user=user_factory(name='unrelated'), score=1),
+ score_factory(post=post3, user=auth_user, score=-1),
+ post1, post2, post3,
+ ])
+ verify_unpaged('special:liked', [1])
+ verify_unpaged('-special:liked', [2, 3])
+
+def test_own_disliked(
+ auth_executor, post_factory, score_factory, user_factory, verify_unpaged):
+ auth_user = auth_executor()
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ db.session.add_all([
+ score_factory(post=post1, user=auth_user, score=-1),
+ score_factory(post=post2, user=user_factory(name='unrelated'), score=-1),
+ score_factory(post=post3, user=auth_user, score=1),
+ post1, post2, post3,
+ ])
+ verify_unpaged('special:disliked', [1])
+ verify_unpaged('-special:disliked', [2, 3])
+
+def test_own_fav(
+ auth_executor, post_factory, fav_factory, user_factory, verify_unpaged):
+ auth_user = auth_executor()
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ db.session.add_all([
+ fav_factory(post=post1, user=auth_user),
+ fav_factory(post=post2, user=user_factory(name='unrelated')),
+ post1, post2,
+ ])
+ verify_unpaged('special:fav', [1])
+ verify_unpaged('-special:fav', [2])
+
+def test_tumbleweed(
+ executor,
+ post_factory,
+ fav_factory,
+ comment_factory,
+ score_factory,
+ verify_unpaged):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ post3 = post_factory(id=3)
+ post4 = post_factory(id=4)
+ db.session.add_all([
+ comment_factory(post=post1),
+ score_factory(post=post2),
+ fav_factory(post=post3),
+ post1, post2, post3, post4,
+ ])
+ verify_unpaged('special:tumbleweed', [4])
+ verify_unpaged('-special:tumbleweed', [1, 2, 3])