summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorRuin0x11 <ipickering2@gmail.com>2020-05-04 19:12:54 -0700
committerRuin0x11 <ipickering2@gmail.com>2020-05-04 19:12:54 -0700
commit5ca21f9e7f5447a2c7841c1317dbe8df01b04e97 (patch)
treeb807a323d84653c2a1af707436bd002973cc132f
parent6b8e3f251f69c9d7d560148100cdb8df941a81c2 (diff)
Add pool tests
-rw-r--r--README.md1
-rw-r--r--server/szurubooru/api/pool_api.py1
-rw-r--r--server/szurubooru/func/pools.py31
-rw-r--r--server/szurubooru/model/pool.py2
-rw-r--r--server/szurubooru/model/post.py1
-rw-r--r--server/szurubooru/tests/api/test_pool_category_creating.py58
-rw-r--r--server/szurubooru/tests/api/test_pool_category_deleting.py76
-rw-r--r--server/szurubooru/tests/api/test_pool_category_retrieving.py56
-rw-r--r--server/szurubooru/tests/api/test_pool_category_updating.py108
-rw-r--r--server/szurubooru/tests/api/test_pool_creating.py82
-rw-r--r--server/szurubooru/tests/api/test_pool_deleting.py61
-rw-r--r--server/szurubooru/tests/api/test_pool_merging.py98
-rw-r--r--server/szurubooru/tests/api/test_pool_retrieving.py72
-rw-r--r--server/szurubooru/tests/api/test_pool_updating.py128
-rw-r--r--server/szurubooru/tests/conftest.py46
-rw-r--r--server/szurubooru/tests/func/test_posts.py43
-rw-r--r--server/szurubooru/tests/model/test_pool.py97
-rw-r--r--server/szurubooru/tests/search/configs/test_pool_search_config.py338
18 files changed, 1292 insertions, 7 deletions
diff --git a/README.md b/README.md
index 98c0c0c..47e6253 100644
--- a/README.md
+++ b/README.md
@@ -20,6 +20,7 @@ scrubbing](http://sjp.pwn.pl/sjp/;2527372). It is pronounced as *shoorubooru*.
- Tag suggestions
- Tag implications (adding a tag automatically adds another)
- Tag aliases
+- Pools and pool categories
- Duplicate detection
- Post rating and favoriting; comment rating
- Polished UI
diff --git a/server/szurubooru/api/pool_api.py b/server/szurubooru/api/pool_api.py
index 69b9951..3d4f74c 100644
--- a/server/szurubooru/api/pool_api.py
+++ b/server/szurubooru/api/pool_api.py
@@ -34,6 +34,7 @@ def create_pool(
posts = ctx.get_param_as_int_list('posts', default=[])
pool = pools.create_pool(names, category, posts)
+ pool.last_edit_time = datetime.utcnow()
pools.update_pool_description(pool, description)
ctx.session.add(pool)
ctx.session.flush()
diff --git a/server/szurubooru/func/pools.py b/server/szurubooru/func/pools.py
index 854650a..27574ee 100644
--- a/server/szurubooru/func/pools.py
+++ b/server/szurubooru/func/pools.py
@@ -38,6 +38,10 @@ class InvalidPoolRelationError(errors.ValidationError):
pass
+class InvalidPoolNonexistentPostError(errors.ValidationError):
+ pass
+
+
def _verify_name_validity(name: str) -> None:
if util.value_exceeds_column_size(name, model.PoolName.name):
raise InvalidPoolNameError('Name is too long.')
@@ -63,9 +67,15 @@ def _check_name_intersection(
return len(set(names1).intersection(names2)) > 0
-def _check_post_duplication(post_ids: List[int]) -> bool:
- return len(post_ids) != len(set(post_ids))
-
+def _duplicates(a: List[int]) -> List[int]:
+ seen = set()
+ dupes = []
+ for x in a:
+ if x not in seen:
+ seen.add(x)
+ else:
+ dupes.append(x)
+ return dupes
def sort_pools(pools: List[model.Pool]) -> List[model.Pool]:
default_category_name = pool_categories.get_default_category_name()
@@ -294,8 +304,17 @@ def update_pool_description(pool: model.Pool, description: str) -> None:
def update_pool_posts(pool: model.Pool, post_ids: List[int]) -> None:
assert pool
- if _check_post_duplication(post_ids):
- raise InvalidPoolDuplicateError('Duplicate post in pool.')
+ dupes = _duplicates(post_ids)
+ if len(dupes) > 0:
+ print(str(dupes))
+ print(str(post_ids))
+ dupes = ', '.join(list(str(x) for x in dupes))
+ raise InvalidPoolDuplicateError('Duplicate post(s) in pool: ' + dupes)
+ ret = posts.get_posts_by_ids(post_ids)
+ if len(post_ids) != len(ret):
+ missing = set(post_ids) - set(post.post_id for post in ret)
+ missing = ', '.join(list(str(x) for x in missing))
+ raise InvalidPoolNonexistentPostError('The following posts do not exist: ' + missing)
pool.posts.clear()
- for post in posts.get_posts_by_ids(post_ids):
+ for post in ret:
pool.posts.append(post)
diff --git a/server/szurubooru/model/pool.py b/server/szurubooru/model/pool.py
index 8f7c605..70595d9 100644
--- a/server/szurubooru/model/pool.py
+++ b/server/szurubooru/model/pool.py
@@ -70,9 +70,9 @@ class Pool(Base):
order_by='PoolName.order')
_posts = sa.orm.relationship(
'PoolPost',
- back_populates='pool',
cascade='all,delete-orphan',
lazy='joined',
+ back_populates='pool',
order_by='PoolPost.order',
collection_class=ordering_list('order'))
posts = association_proxy('_posts', 'post')
diff --git a/server/szurubooru/model/post.py b/server/szurubooru/model/post.py
index 56eaefa..11da866 100644
--- a/server/szurubooru/model/post.py
+++ b/server/szurubooru/model/post.py
@@ -229,6 +229,7 @@ class Post(Base):
comments = sa.orm.relationship('Comment', cascade='all, delete-orphan')
_pools = sa.orm.relationship(
'PoolPost',
+ cascade='all,delete-orphan',
lazy='select',
order_by='PoolPost.order',
back_populates='post')
diff --git a/server/szurubooru/tests/api/test_pool_category_creating.py b/server/szurubooru/tests/api/test_pool_category_creating.py
new file mode 100644
index 0000000..4accb92
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_category_creating.py
@@ -0,0 +1,58 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pool_categories, snapshots
+
+
+def _update_category_name(category, name):
+ category.name = name
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'privileges': {'pool_categories:create': model.User.RANK_REGULAR},
+ })
+
+
+def test_creating_category(
+ pool_category_factory, user_factory, context_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ category = pool_category_factory(name='meta')
+ db.session.add(category)
+
+ with patch('szurubooru.func.pool_categories.create_category'), \
+ patch('szurubooru.func.pool_categories.serialize_category'), \
+ patch('szurubooru.func.pool_categories.update_category_name'), \
+ patch('szurubooru.func.snapshots.create'):
+ pool_categories.create_category.return_value = category
+ pool_categories.update_category_name.side_effect = _update_category_name
+ pool_categories.serialize_category.return_value = 'serialized category'
+ result = api.pool_category_api.create_pool_category(
+ context_factory(
+ params={'name': 'meta', 'color': 'black'}, user=auth_user))
+ assert result == 'serialized category'
+ pool_categories.create_category.assert_called_once_with('meta', 'black')
+ snapshots.create.assert_called_once_with(category, auth_user)
+
+
+@pytest.mark.parametrize('field', ['name', 'color'])
+def test_trying_to_omit_mandatory_field(user_factory, context_factory, field):
+ params = {
+ 'name': 'meta',
+ 'color': 'black',
+ }
+ del params[field]
+ with pytest.raises(errors.ValidationError):
+ api.pool_category_api.create_pool_category(
+ context_factory(
+ params=params,
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+
+
+def test_trying_to_create_without_privileges(user_factory, context_factory):
+ with pytest.raises(errors.AuthError):
+ api.pool_category_api.create_pool_category(
+ context_factory(
+ params={'name': 'meta', 'color': 'black'},
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)))
diff --git a/server/szurubooru/tests/api/test_pool_category_deleting.py b/server/szurubooru/tests/api/test_pool_category_deleting.py
new file mode 100644
index 0000000..72853ac
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_category_deleting.py
@@ -0,0 +1,76 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pool_categories, snapshots
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'privileges': {'pool_categories:delete': model.User.RANK_REGULAR},
+ })
+
+
+def test_deleting(user_factory, pool_category_factory, context_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ category = pool_category_factory(name='category')
+ db.session.add(pool_category_factory(name='root'))
+ db.session.add(category)
+ db.session.flush()
+ with patch('szurubooru.func.snapshots.delete'):
+ result = api.pool_category_api.delete_pool_category(
+ context_factory(params={'version': 1}, user=auth_user),
+ {'category_name': 'category'})
+ assert result == {}
+ assert db.session.query(model.PoolCategory).count() == 1
+ assert db.session.query(model.PoolCategory).one().name == 'root'
+ snapshots.delete.assert_called_once_with(category, auth_user)
+
+
+def test_trying_to_delete_used(
+ user_factory, pool_category_factory, pool_factory, context_factory):
+ category = pool_category_factory(name='category')
+ db.session.add(category)
+ db.session.flush()
+ pool = pool_factory(names=['pool'], category=category)
+ db.session.add(pool)
+ db.session.commit()
+ with pytest.raises(pool_categories.PoolCategoryIsInUseError):
+ api.pool_category_api.delete_pool_category(
+ context_factory(
+ params={'version': 1},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'category'})
+ assert db.session.query(model.PoolCategory).count() == 1
+
+
+def test_trying_to_delete_last(
+ user_factory, pool_category_factory, context_factory):
+ db.session.add(pool_category_factory(name='root'))
+ db.session.commit()
+ with pytest.raises(pool_categories.PoolCategoryIsInUseError):
+ api.pool_category_api.delete_pool_category(
+ context_factory(
+ params={'version': 1},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'root'})
+
+
+def test_trying_to_delete_non_existing(user_factory, context_factory):
+ with pytest.raises(pool_categories.PoolCategoryNotFoundError):
+ api.pool_category_api.delete_pool_category(
+ context_factory(user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'bad'})
+
+
+def test_trying_to_delete_without_privileges(
+ user_factory, pool_category_factory, context_factory):
+ db.session.add(pool_category_factory(name='category'))
+ db.session.commit()
+ with pytest.raises(errors.AuthError):
+ api.pool_category_api.delete_pool_category(
+ context_factory(
+ params={'version': 1},
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)),
+ {'category_name': 'category'})
+ assert db.session.query(model.PoolCategory).count() == 1
diff --git a/server/szurubooru/tests/api/test_pool_category_retrieving.py b/server/szurubooru/tests/api/test_pool_category_retrieving.py
new file mode 100644
index 0000000..4a467c0
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_category_retrieving.py
@@ -0,0 +1,56 @@
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pool_categories
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'privileges': {
+ 'pool_categories:list': model.User.RANK_REGULAR,
+ 'pool_categories:view': model.User.RANK_REGULAR,
+ },
+ })
+
+
+def test_retrieving_multiple(
+ user_factory, pool_category_factory, context_factory):
+ db.session.add_all([
+ pool_category_factory(name='c1'),
+ pool_category_factory(name='c2'),
+ ])
+ db.session.flush()
+ result = api.pool_category_api.get_pool_categories(
+ context_factory(user=user_factory(rank=model.User.RANK_REGULAR)))
+ assert [cat['name'] for cat in result['results']] == ['c1', 'c2']
+
+
+def test_retrieving_single(
+ user_factory, pool_category_factory, context_factory):
+ db.session.add(pool_category_factory(name='cat'))
+ db.session.flush()
+ result = api.pool_category_api.get_pool_category(
+ context_factory(user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'cat'})
+ assert result == {
+ 'name': 'cat',
+ 'color': 'dummy',
+ 'usages': 0,
+ 'default': False,
+ 'version': 1,
+ }
+
+
+def test_trying_to_retrieve_single_non_existing(user_factory, context_factory):
+ with pytest.raises(pool_categories.PoolCategoryNotFoundError):
+ api.pool_category_api.get_pool_category(
+ context_factory(user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': '-'})
+
+
+def test_trying_to_retrieve_single_without_privileges(
+ user_factory, context_factory):
+ with pytest.raises(errors.AuthError):
+ api.pool_category_api.get_pool_category(
+ context_factory(user=user_factory(rank=model.User.RANK_ANONYMOUS)),
+ {'category_name': '-'})
diff --git a/server/szurubooru/tests/api/test_pool_category_updating.py b/server/szurubooru/tests/api/test_pool_category_updating.py
new file mode 100644
index 0000000..9c9f743
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_category_updating.py
@@ -0,0 +1,108 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pool_categories, snapshots
+
+
+def _update_category_name(category, name):
+ category.name = name
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'privileges': {
+ 'pool_categories:edit:name': model.User.RANK_REGULAR,
+ 'pool_categories:edit:color': model.User.RANK_REGULAR,
+ 'pool_categories:set_default': model.User.RANK_REGULAR,
+ },
+ })
+
+
+def test_simple_updating(user_factory, pool_category_factory, context_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ category = pool_category_factory(name='name', color='black')
+ db.session.add(category)
+ db.session.flush()
+ with patch('szurubooru.func.pool_categories.serialize_category'), \
+ patch('szurubooru.func.pool_categories.update_category_name'), \
+ patch('szurubooru.func.pool_categories.update_category_color'), \
+ patch('szurubooru.func.snapshots.modify'):
+ pool_categories.update_category_name.side_effect = _update_category_name
+ pool_categories.serialize_category.return_value = 'serialized category'
+ result = api.pool_category_api.update_pool_category(
+ context_factory(
+ params={'name': 'changed', 'color': 'white', 'version': 1},
+ user=auth_user),
+ {'category_name': 'name'})
+ assert result == 'serialized category'
+ pool_categories.update_category_name.assert_called_once_with(
+ category, 'changed')
+ pool_categories.update_category_color.assert_called_once_with(
+ category, 'white')
+ snapshots.modify.assert_called_once_with(category, auth_user)
+
+
+@pytest.mark.parametrize('field', ['name', 'color'])
+def test_omitting_optional_field(
+ user_factory, pool_category_factory, context_factory, field):
+ db.session.add(pool_category_factory(name='name', color='black'))
+ db.session.commit()
+ params = {
+ 'name': 'changed',
+ 'color': 'white',
+ }
+ del params[field]
+ with patch('szurubooru.func.pool_categories.serialize_category'), \
+ patch('szurubooru.func.pool_categories.update_category_name'):
+ api.pool_category_api.update_pool_category(
+ context_factory(
+ params={**params, **{'version': 1}},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'name'})
+
+
+def test_trying_to_update_non_existing(user_factory, context_factory):
+ with pytest.raises(pool_categories.PoolCategoryNotFoundError):
+ api.pool_category_api.update_pool_category(
+ context_factory(
+ params={'name': ['dummy']},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'bad'})
+
+
+@pytest.mark.parametrize('params', [
+ {'name': 'whatever'},
+ {'color': 'whatever'},
+])
+def test_trying_to_update_without_privileges(
+ user_factory, pool_category_factory, context_factory, params):
+ db.session.add(pool_category_factory(name='dummy'))
+ db.session.commit()
+ with pytest.raises(errors.AuthError):
+ api.pool_category_api.update_pool_category(
+ context_factory(
+ params={**params, **{'version': 1}},
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)),
+ {'category_name': 'dummy'})
+
+
+def test_set_as_default(user_factory, pool_category_factory, context_factory):
+ category = pool_category_factory(name='name', color='black')
+ db.session.add(category)
+ db.session.commit()
+ with patch('szurubooru.func.pool_categories.serialize_category'), \
+ patch('szurubooru.func.pool_categories.set_default_category'):
+ pool_categories.update_category_name.side_effect = _update_category_name
+ pool_categories.serialize_category.return_value = 'serialized category'
+ result = api.pool_category_api.set_pool_category_as_default(
+ context_factory(
+ params={
+ 'name': 'changed',
+ 'color': 'white',
+ 'version': 1,
+ },
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'category_name': 'name'})
+ assert result == 'serialized category'
+ pool_categories.set_default_category.assert_called_once_with(category)
diff --git a/server/szurubooru/tests/api/test_pool_creating.py b/server/szurubooru/tests/api/test_pool_creating.py
new file mode 100644
index 0000000..5eeb2e2
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_creating.py
@@ -0,0 +1,82 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, model, errors
+from szurubooru.func import pools, posts, snapshots
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({'privileges': {'pools:create': model.User.RANK_REGULAR}})
+
+
+def test_creating_simple_pools(pool_factory, user_factory, context_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ pool = pool_factory()
+ with patch('szurubooru.func.pools.create_pool'), \
+ patch('szurubooru.func.pools.get_or_create_pools_by_names'), \
+ patch('szurubooru.func.pools.serialize_pool'), \
+ patch('szurubooru.func.snapshots.create'):
+ posts.get_posts_by_ids.return_value = ([], [])
+ pools.create_pool.return_value = pool
+ pools.serialize_pool.return_value = 'serialized pool'
+ result = api.pool_api.create_pool(
+ context_factory(
+ params={
+ 'names': ['pool1', 'pool2'],
+ 'category': 'default',
+ 'description': 'desc',
+ 'posts': [1, 2],
+ },
+ user=auth_user))
+ assert result == 'serialized pool'
+ pools.create_pool.assert_called_once_with(
+ ['pool1', 'pool2'], 'default', [1, 2])
+ snapshots.create.assert_called_once_with(pool, auth_user)
+
+
+@pytest.mark.parametrize('field', ['names', 'category'])
+def test_trying_to_omit_mandatory_field(user_factory, context_factory, field):
+ params = {
+ 'names': ['pool1', 'pool2'],
+ 'category': 'default',
+ 'description': 'desc',
+ 'posts': [],
+ }
+ del params[field]
+ with pytest.raises(errors.ValidationError):
+ api.pool_api.create_pool(
+ context_factory(
+ params=params,
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+
+
+@pytest.mark.parametrize('field', ['description', 'posts'])
+def test_omitting_optional_field(
+ pool_factory, user_factory, context_factory, field):
+ params = {
+ 'names': ['pool1', 'pool2'],
+ 'category': 'default',
+ 'description': 'desc',
+ 'posts': [],
+ }
+ del params[field]
+ with patch('szurubooru.func.pools.create_pool'), \
+ patch('szurubooru.func.pools.serialize_pool'):
+ pools.create_pool.return_value = pool_factory()
+ api.pool_api.create_pool(
+ context_factory(
+ params=params,
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+
+
+def test_trying_to_create_pool_without_privileges(
+ user_factory, context_factory):
+ with pytest.raises(errors.AuthError):
+ api.pool_api.create_pool(
+ context_factory(
+ params={
+ 'names': ['pool'],
+ 'category': 'default',
+ 'posts': [],
+ },
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)))
diff --git a/server/szurubooru/tests/api/test_pool_deleting.py b/server/szurubooru/tests/api/test_pool_deleting.py
new file mode 100644
index 0000000..e29656d
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_deleting.py
@@ -0,0 +1,61 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pools, snapshots
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({'privileges': {'pools:delete': model.User.RANK_REGULAR}})
+
+
+def test_deleting(user_factory, pool_factory, context_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ pool = pool_factory(id=1)
+ db.session.add(pool)
+ db.session.commit()
+ with patch('szurubooru.func.snapshots.delete'):
+ result = api.pool_api.delete_pool(
+ context_factory(params={'version': 1}, user=auth_user),
+ {'pool_id': 1})
+ assert result == {}
+ assert db.session.query(model.Pool).count() == 0
+ snapshots.delete.assert_called_once_with(pool, auth_user)
+
+
+def test_deleting_used(
+ user_factory, pool_factory, context_factory, post_factory):
+ pool = pool_factory(id=1)
+ post = post_factory(id=1)
+ pool.posts.append(post)
+ db.session.add_all([pool, post])
+ db.session.commit()
+ api.pool_api.delete_pool(
+ context_factory(
+ params={'version': 1},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 1})
+ db.session.refresh(post)
+ assert db.session.query(model.Pool).count() == 0
+ assert db.session.query(model.PoolPost).count() == 0
+ assert post.pools == []
+
+
+def test_trying_to_delete_non_existing(user_factory, context_factory):
+ with pytest.raises(pools.PoolNotFoundError):
+ api.pool_api.delete_pool(
+ context_factory(user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 9999})
+
+
+def test_trying_to_delete_without_privileges(
+ user_factory, pool_factory, context_factory):
+ db.session.add(pool_factory(id=1))
+ db.session.commit()
+ with pytest.raises(errors.AuthError):
+ api.pool_api.delete_pool(
+ context_factory(
+ params={'version': 1},
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)),
+ {'pool_id': 1})
+ assert db.session.query(model.Pool).count() == 1
diff --git a/server/szurubooru/tests/api/test_pool_merging.py b/server/szurubooru/tests/api/test_pool_merging.py
new file mode 100644
index 0000000..dc462d2
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_merging.py
@@ -0,0 +1,98 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pools, snapshots
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({'privileges': {'pools:merge': model.User.RANK_REGULAR}})
+
+
+def test_merging(user_factory, pool_factory, context_factory, post_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ source_pool = pool_factory(id=1)
+ target_pool = pool_factory(id=2)
+ db.session.add_all([source_pool, target_pool])
+ db.session.flush()
+ assert source_pool.post_count == 0
+ assert target_pool.post_count == 0
+ post = post_factory(id=1)
+ source_pool.posts = [post]
+ db.session.add(post)
+ db.session.commit()
+ assert source_pool.post_count == 1
+ assert target_pool.post_count == 0
+ with patch('szurubooru.func.pools.serialize_pool'), \
+ patch('szurubooru.func.pools.merge_pools'), \
+ patch('szurubooru.func.snapshots.merge'):
+ api.pool_api.merge_pools(
+ context_factory(
+ params={
+ 'removeVersion': 1,
+ 'mergeToVersion': 1,
+ 'remove': 1,
+ 'mergeTo': 2,
+ },
+ user=auth_user))
+ pools.merge_pools.called_once_with(source_pool, target_pool)
+ snapshots.merge.assert_called_once_with(
+ source_pool, target_pool, auth_user)
+
+
+@pytest.mark.parametrize(
+ 'field', ['remove', 'mergeTo', 'removeVersion', 'mergeToVersion'])
+def test_trying_to_omit_mandatory_field(
+ user_factory, pool_factory, context_factory, field):
+ db.session.add_all([
+ pool_factory(id=1),
+ pool_factory(id=2),
+ ])
+ db.session.commit()
+ params = {
+ 'removeVersion': 1,
+ 'mergeToVersion': 1,
+ 'remove': 1,
+ 'mergeTo': 2,
+ }
+ del params[field]
+ with pytest.raises(errors.ValidationError):
+ api.pool_api.merge_pools(
+ context_factory(
+ params=params,
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+
+
+def test_trying_to_merge_non_existing(
+ user_factory, pool_factory, context_factory):
+ db.session.add(pool_factory(id=1))
+ db.session.commit()
+ with pytest.raises(pools.PoolNotFoundError):
+ api.pool_api.merge_pools(
+ context_factory(
+ params={'remove': 1, 'mergeTo': 9999},
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+ with pytest.raises(pools.PoolNotFoundError):
+ api.pool_api.merge_pools(
+ context_factory(
+ params={'remove': 9999, 'mergeTo': 1},
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+
+
+def test_trying_to_merge_without_privileges(
+ user_factory, pool_factory, context_factory):
+ db.session.add_all([
+ pool_factory(id=1),
+ pool_factory(id=2),
+ ])
+ db.session.commit()
+ with pytest.raises(errors.AuthError):
+ api.pool_api.merge_pools(
+ context_factory(
+ params={
+ 'removeVersion': 1,
+ 'mergeToVersion': 1,
+ 'remove': 1,
+ 'mergeTo': 2,
+ },
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)))
diff --git a/server/szurubooru/tests/api/test_pool_retrieving.py b/server/szurubooru/tests/api/test_pool_retrieving.py
new file mode 100644
index 0000000..e48565a
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_retrieving.py
@@ -0,0 +1,72 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pools
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'privileges': {
+ 'pools:list': model.User.RANK_REGULAR,
+ 'pools:view': model.User.RANK_REGULAR,
+ },
+ })
+
+
+def test_retrieving_multiple(user_factory, pool_factory, context_factory):
+ pool1 = pool_factory(id=1)
+ pool2 = pool_factory(id=2)
+ db.session.add_all([pool2, pool1])
+ db.session.flush()
+ with patch('szurubooru.func.pools.serialize_pool'):
+ pools.serialize_pool.return_value = 'serialized pool'
+ result = api.pool_api.get_pools(
+ context_factory(
+ params={'query': '', 'offset': 0},
+ user=user_factory(rank=model.User.RANK_REGULAR)))
+ assert result == {
+ 'query': '',
+ 'offset': 0,
+ 'limit': 100,
+ 'total': 2,
+ 'results': ['serialized pool', 'serialized pool'],
+ }
+
+
+def test_trying_to_retrieve_multiple_without_privileges(
+ user_factory, context_factory):
+ with pytest.raises(errors.AuthError):
+ api.pool_api.get_pools(
+ context_factory(
+ params={'query': '', 'offset': 0},
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)))
+
+
+def test_retrieving_single(user_factory, pool_factory, context_factory):
+ db.session.add(pool_factory(id=1))
+ db.session.flush()
+ with patch('szurubooru.func.pools.serialize_pool'):
+ pools.serialize_pool.return_value = 'serialized pool'
+ result = api.pool_api.get_pool(
+ context_factory(
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 1})
+ assert result == 'serialized pool'
+
+
+def test_trying_to_retrieve_single_non_existing(user_factory, context_factory):
+ with pytest.raises(pools.PoolNotFoundError):
+ api.pool_api.get_pool(
+ context_factory(
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 1})
+
+
+def test_trying_to_retrieve_single_without_privileges(
+ user_factory, context_factory):
+ with pytest.raises(errors.AuthError):
+ api.pool_api.get_pool(
+ context_factory(
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)),
+ {'pool_id': 1})
diff --git a/server/szurubooru/tests/api/test_pool_updating.py b/server/szurubooru/tests/api/test_pool_updating.py
new file mode 100644
index 0000000..52eddc9
--- /dev/null
+++ b/server/szurubooru/tests/api/test_pool_updating.py
@@ -0,0 +1,128 @@
+from unittest.mock import patch
+import pytest
+from szurubooru import api, db, model, errors
+from szurubooru.func import pools, posts, snapshots
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'privileges': {
+ 'pools:create': model.User.RANK_REGULAR,
+ 'pools:edit:names': model.User.RANK_REGULAR,
+ 'pools:edit:category': model.User.RANK_REGULAR,
+ 'pools:edit:description': model.User.RANK_REGULAR,
+ 'pools:edit:posts': model.User.RANK_REGULAR,
+ },
+ })
+
+
+def test_simple_updating(user_factory, pool_factory, context_factory):
+ auth_user = user_factory(rank=model.User.RANK_REGULAR)
+ pool = pool_factory(id=1, names=['pool1', 'pool2'])
+ db.session.add(pool)
+ db.session.commit()
+ with patch('szurubooru.func.pools.create_pool'), \
+ patch('szurubooru.func.posts.get_posts_by_ids'), \
+ patch('szurubooru.func.pools.update_pool_names'), \
+ patch('szurubooru.func.pools.update_pool_category_name'), \
+ patch('szurubooru.func.pools.update_pool_description'), \
+ patch('szurubooru.func.pools.update_pool_posts'), \
+ patch('szurubooru.func.pools.serialize_pool'), \
+ patch('szurubooru.func.snapshots.modify'):
+ posts.get_posts_by_ids.return_value = ([], [])
+ pools.serialize_pool.return_value = 'serialized pool'
+ result = api.pool_api.update_pool(
+ context_factory(
+ params={
+ 'version': 1,
+ 'names': ['pool3'],
+ 'category': 'series',
+ 'description': 'desc',
+ 'posts': [1, 2]
+ },
+ user=auth_user),
+ {'pool_id': 1})
+ assert result == 'serialized pool'
+ pools.create_pool.assert_not_called()
+ pools.update_pool_names.assert_called_once_with(pool, ['pool3'])
+ pools.update_pool_category_name.assert_called_once_with(pool, 'series')
+ pools.update_pool_description.assert_called_once_with(pool, 'desc')
+ pools.update_pool_posts.assert_called_once_with(pool, [1, 2])
+ pools.serialize_pool.assert_called_once_with(pool, options=[])
+ snapshots.modify.assert_called_once_with(pool, auth_user)
+
+
+@pytest.mark.parametrize(
+ 'field', [
+ 'names',
+ 'category',
+ 'description',
+ 'posts',
+ ])
+def test_omitting_optional_field(
+ user_factory, pool_factory, context_factory, field):
+ db.session.add(pool_factory(id=1))
+ db.session.commit()
+ params = {
+ 'names': ['pool1', 'pool2'],
+ 'category': 'default',
+ 'description': 'desc',
+ 'posts': [],
+ }
+ del params[field]
+ with patch('szurubooru.func.pools.create_pool'), \
+ patch('szurubooru.func.pools.update_pool_names'), \
+ patch('szurubooru.func.pools.update_pool_category_name'), \
+ patch('szurubooru.func.pools.serialize_pool'):
+ api.pool_api.update_pool(
+ context_factory(
+ params={**params, **{'version': 1}},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 1})
+
+
+def test_trying_to_update_non_existing(user_factory, context_factory):
+ with pytest.raises(pools.PoolNotFoundError):
+ api.pool_api.update_pool(
+ context_factory(
+ params={'names': ['dummy']},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 9999})
+
+
+@pytest.mark.parametrize('params', [
+ {'names': ['whatever']},
+ {'category': 'whatever'},
+ {'posts': [1]},
+])
+def test_trying_to_update_without_privileges(
+ user_factory, pool_factory, context_factory, params):
+ db.session.add(pool_factory(id=1))
+ db.session.commit()
+ with pytest.raises(errors.AuthError):
+ api.pool_api.update_pool(
+ context_factory(
+ params={**params, **{'version': 1}},
+ user=user_factory(rank=model.User.RANK_ANONYMOUS)),
+ {'pool_id': 1})
+
+
+def test_trying_to_create_pools_without_privileges(
+ config_injector, context_factory, pool_factory, user_factory):
+ pool = pool_factory(id=1)
+ db.session.add(pool)
+ db.session.commit()
+ config_injector({'privileges': {
+ 'pools:create': model.User.RANK_ADMINISTRATOR,
+ 'pools:edit:posts': model.User.RANK_REGULAR,
+ },
+ 'delete_source_files': False})
+ with patch('szurubooru.func.posts.get_posts_by_ids'):
+ posts.get_posts_by_ids.return_value = ([], ['new-post'])
+ with pytest.raises(errors.AuthError):
+ api.pool_api.create_pool(
+ context_factory(
+ params={'posts': [1, 2], 'version': 1},
+ user=user_factory(rank=model.User.RANK_REGULAR)),
+ {'pool_id': 1})
diff --git a/server/szurubooru/tests/conftest.py b/server/szurubooru/tests/conftest.py
index a131fec..249df1a 100644
--- a/server/szurubooru/tests/conftest.py
+++ b/server/szurubooru/tests/conftest.py
@@ -202,6 +202,52 @@ def post_favorite_factory(user_factory, post_factory):
@pytest.fixture
+def pool_category_factory():
+ def factory(name=None, color='dummy', default=False):
+ category = model.PoolCategory()
+ category.name = name or get_unique_name()
+ category.color = color
+ category.default = default
+ return category
+ return factory
+
+
+@pytest.fixture
+def pool_factory():
+ def factory(id=None, names=None, description=None, category=None, time=None):
+ if not category:
+ category = model.PoolCategory(get_unique_name())
+ db.session.add(category)
+ pool = model.Pool()
+ pool.pool_id = id
+ pool.names = []
+ for i, name in enumerate(names or [get_unique_name()]):
+ pool.names.append(model.PoolName(name, i))
+ pool.description = description
+ pool.category = category
+ pool.creation_time = time or datetime(1996, 1, 1)
+ return pool
+ return factory
+
+
+@pytest.fixture
+def pool_post_factory(pool_factory, post_factory):
+ def factory(pool=None, post=None, order=None):
+ if not pool:
+ pool = pool_factory()
+ db.session.add(pool)
+ if not post:
+ post = post_factory()
+ db.session.add(post)
+ pool_post = model.PoolPost(post)
+ pool_post.pool = pool
+ pool_post.post = post
+ pool_post.order = order or 0
+ return pool_post
+ return factory
+
+
+@pytest.fixture
def read_asset():
def get(path):
path = os.path.join(os.path.dirname(__file__), 'assets', path)
diff --git a/server/szurubooru/tests/func/test_posts.py b/server/szurubooru/tests/func/test_posts.py
index 097136c..e785ea5 100644
--- a/server/szurubooru/tests/func/test_posts.py
+++ b/server/szurubooru/tests/func/test_posts.py
@@ -79,6 +79,8 @@ def test_serialize_post(
comment_factory,
tag_factory,
tag_category_factory,
+ pool_factory,
+ pool_category_factory,
config_injector):
config_injector({'data_url': 'http://example.com/', 'secret': 'test'})
with patch('szurubooru.func.comments.serialize_comment'), \
@@ -150,6 +152,23 @@ def test_serialize_post(
time=datetime(1800, 1, 1))])
db.session.flush()
+ pool1 = pool_factory(id=1,
+ names=['pool1', 'pool2'],
+ description='desc',
+ category=pool_category_factory('test-cat1'))
+ pool1.last_edit_time = datetime(1998, 1, 1)
+ pool1.posts.append(post)
+
+ pool2 = pool_factory(id=2,
+ names=['pool3'],
+ description='desc2',
+ category=pool_category_factory('test-cat2'))
+ pool2.last_edit_time = datetime(1998, 1, 1)
+ pool2.posts.append(post)
+
+ db.session.add_all([pool1, pool2])
+ db.session.flush()
+
result = posts.serialize_post(post, auth_user)
result['tags'].sort(key=lambda tag: tag['names'][0])
@@ -183,6 +202,30 @@ def test_serialize_post(
],
'relations': [],
'notes': [],
+ 'pools': [
+ {
+ 'id': 1,
+ 'names': ['pool1', 'pool2'],
+ 'description': 'desc',
+ 'category': 'test-cat1',
+ 'postCount': 1,
+ 'posts': [{'id': 1}],
+ 'version': 1,
+ 'creationTime': datetime(1996, 1, 1),
+ 'lastEditTime': datetime(1998, 1, 1),
+ },
+ {
+ 'id': 2,
+ 'names': ['pool3'],
+ 'description': 'desc2',
+ 'category': 'test-cat2',
+ 'postCount': 1,
+ 'posts': [{'id': 1}],
+ 'version': 1,
+ 'creationTime': datetime(1996, 1, 1),
+ 'lastEditTime': datetime(1998, 1, 1),
+ }
+ ],
'user': 'post author',
'score': 1,
'ownFavorite': False,
diff --git a/server/szurubooru/tests/model/test_pool.py b/server/szurubooru/tests/model/test_pool.py
new file mode 100644
index 0000000..589e9ba
--- /dev/null
+++ b/server/szurubooru/tests/model/test_pool.py
@@ -0,0 +1,97 @@
+from datetime import datetime
+import pytest
+from szurubooru import db, model
+
+
+@pytest.fixture(autouse=True)
+def inject_config(config_injector):
+ config_injector({
+ 'delete_source_files': False,
+ 'secret': 'secret',
+ 'data_dir': ''
+ })
+
+
+def test_saving_pool(pool_factory, post_factory):
+ post1 = post_factory()
+ post2 = post_factory()
+ pool = model.Pool()
+ pool.names = [model.PoolName('alias1', 0), model.PoolName('alias2', 1)]
+ pool.posts = []
+ pool.category = model.PoolCategory('category')
+ pool.creation_time = datetime(1997, 1, 1)
+ pool.last_edit_time = datetime(1998, 1, 1)
+ db.session.add_all([pool, post1, post2])
+ db.session.commit()
+
+ assert pool.pool_id is not None
+ pool.posts.append(post1)
+ pool.posts.append(post2)
+ db.session.commit()
+
+ pool = (
+ db.session
+ .query(model.Pool)
+ .join(model.PoolName)
+ .filter(model.PoolName.name == 'alias1')
+ .one())
+ assert [pool_name.name for pool_name in pool.names] == ['alias1', 'alias2']
+ assert pool.category.name == 'category'
+ assert pool.creation_time == datetime(1997, 1, 1)
+ assert pool.last_edit_time == datetime(1998, 1, 1)
+ assert [post.post_id for post in pool.posts] == [1, 2]
+
+
+def test_cascade_deletions(pool_factory, post_factory):
+ post1 = post_factory()
+ post2 = post_factory()
+ pool = model.Pool()
+ pool.names = [model.PoolName('alias1', 0), model.PoolName('alias2', 1)]
+ pool.posts = []
+ pool.category = model.PoolCategory('category')
+ pool.creation_time = datetime(1997, 1, 1)
+ pool.last_edit_time = datetime(1998, 1, 1)
+ db.session.add_all([pool, post1, post2])
+ db.session.commit()
+
+ assert pool.pool_id is not None
+ pool.posts.append(post1)
+ pool.posts.append(post2)
+ db.session.commit()
+
+ db.session.delete(pool)
+ db.session.commit()
+ assert db.session.query(model.Pool).count() == 0
+ assert db.session.query(model.PoolName).count() == 0
+ assert db.session.query(model.PoolPost).count() == 0
+ assert db.session.query(model.PoolCategory).count() == 1
+ assert db.session.query(model.Post).count() == 2
+
+
+def test_tracking_post_count(post_factory, pool_factory):
+ pool1 = pool_factory()
+ pool2 = pool_factory()
+ post1 = post_factory()
+ post2 = post_factory()
+ db.session.add_all([pool1, pool2, post1, post2])
+ db.session.flush()
+ assert pool1.pool_id is not None
+ assert pool2.pool_id is not None
+ pool1.posts.append(post1)
+ pool2.posts.append(post1)
+ pool2.posts.append(post2)
+ db.session.commit()
+ assert len(post1.pools) == 2
+ assert len(post2.pools) == 1
+ assert pool1.post_count == 1
+ assert pool2.post_count == 2
+ db.session.delete(post1)
+ db.session.commit()
+ db.session.refresh(pool1)
+ db.session.refresh(pool2)
+ assert pool1.post_count == 0
+ assert pool2.post_count == 1
+ db.session.delete(post2)
+ db.session.commit()
+ db.session.refresh(pool2)
+ assert pool2.post_count == 0
diff --git a/server/szurubooru/tests/search/configs/test_pool_search_config.py b/server/szurubooru/tests/search/configs/test_pool_search_config.py
new file mode 100644
index 0000000..e013363
--- /dev/null
+++ b/server/szurubooru/tests/search/configs/test_pool_search_config.py
@@ -0,0 +1,338 @@
+# pylint: disable=redefined-outer-name
+from datetime import datetime
+import pytest
+from szurubooru import db, errors, search
+
+
+@pytest.fixture
+def executor():
+ return search.Executor(search.configs.PoolSearchConfig())
+
+
+@pytest.fixture
+def verify_unpaged(executor):
+ def verify(input, expected_pool_names):
+ actual_count, actual_pools = executor.execute(
+ input, offset=0, limit=100)
+ actual_pool_names = [u.names[0].name for u in actual_pools]
+ assert actual_count == len(expected_pool_names)
+ assert actual_pool_names == expected_pool_names
+ return verify
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('', ['t1', 't2']),
+ ('t1', ['t1']),
+ ('t2', ['t2']),
+ ('t1,t2', ['t1', 't2']),
+ ('T1,T2', ['t1', 't2']),
+])
+def test_filter_anonymous(
+ verify_unpaged, pool_factory, input, expected_pool_names):
+ db.session.add(pool_factory(id=1, names=['t1']))
+ db.session.add(pool_factory(id=2, names=['t2']))
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('db_driver,input,expected_pool_names', [
+ (None, ',', None),
+ (None, 't1,', None),
+ (None, 't1,t2', ['t1', 't2']),
+ (None, 't1\\,', []),
+ (None, 'asd..asd', None),
+ (None, 'asd\\..asd', []),
+ (None, 'asd.\\.asd', []),
+ (None, 'asd\\.\\.asd', []),
+ (None, '-', None),
+ (None, '\\-', ['-']),
+ (None, '--', [
+ 't1', 't2', '*', '*asd*', ':', 'asd:asd', '\\', '\\asd', '-asd',
+ ]),
+ (None, '\\--', []),
+ (None, '-\\-', [
+ 't1', 't2', '*', '*asd*', ':', 'asd:asd', '\\', '\\asd', '-asd',
+ ]),
+ (None, '-*', []),
+ (None, '\\-*', ['-', '-asd']),
+ (None, ':', None),
+ (None, '\\:', [':']),
+ (None, '\\:asd', []),
+ (None, '*\\:*', [':', 'asd:asd']),
+ (None, 'asd:asd', None),
+ (None, 'asd\\:asd', ['asd:asd']),
+ (None, '*', [
+ 't1', 't2', '*', '*asd*', ':', 'asd:asd', '\\', '\\asd', '-', '-asd'
+ ]),
+ (None, '\\*', ['*']),
+ (None, '\\', None),
+ (None, '\\asd', None),
+ ('psycopg2', '\\\\', ['\\']),
+ ('psycopg2', '\\\\asd', ['\\asd']),
+])
+def test_escaping(
+ executor, pool_factory, input, expected_pool_names, db_driver):
+ db.session.add_all([
+ pool_factory(id=1, names=['t1']),
+ pool_factory(id=2, names=['t2']),
+ pool_factory(id=3, names=['*']),
+ pool_factory(id=4, names=['*asd*']),
+ pool_factory(id=5, names=[':']),
+ pool_factory(id=6, names=['asd:asd']),
+ pool_factory(id=7, names=['\\']),
+ pool_factory(id=8, names=['\\asd']),
+ pool_factory(id=9, names=['-']),
+ pool_factory(id=10, names=['-asd'])
+ ])
+ db.session.flush()
+
+ if db_driver and db.session.get_bind().driver != db_driver:
+ pytest.xfail()
+ if expected_pool_names is None:
+ with pytest.raises(errors.SearchError):
+ executor.execute(input, offset=0, limit=100)
+ else:
+ actual_count, actual_pools = executor.execute(
+ input, offset=0, limit=100)
+ actual_pool_names = [u.names[0].name for u in actual_pools]
+ assert actual_count == len(expected_pool_names)
+ assert sorted(actual_pool_names) == sorted(expected_pool_names)
+
+
+def test_filter_anonymous_starting_with_colon(verify_unpaged, pool_factory):
+ db.session.add(pool_factory(id=1, names=[':t']))
+ db.session.flush()
+ with pytest.raises(errors.SearchError):
+ verify_unpaged(':t', [':t'])
+ verify_unpaged('\\:t', [':t'])
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('name:pool1', ['pool1']),
+ ('name:pool2', ['pool2']),
+ ('name:none', []),
+ ('name:', []),
+ ('name:*1', ['pool1']),
+ ('name:*2', ['pool2']),
+ ('name:*', ['pool1', 'pool2', 'pool3', 'pool4']),
+ ('name:p*', ['pool1', 'pool2', 'pool3', 'pool4']),
+ ('name:*o*', ['pool1', 'pool2', 'pool3', 'pool4']),
+ ('name:*!*', []),
+ ('name:!*', []),
+ ('name:*!', []),
+ ('-name:pool1', ['pool2', 'pool3', 'pool4']),
+ ('-name:pool2', ['pool1', 'pool3', 'pool4']),
+ ('name:pool1,pool2', ['pool1', 'pool2']),
+ ('-name:pool1,pool3', ['pool2', 'pool4']),
+ ('name:pool4', ['pool4']),
+ ('name:pool5', ['pool4']),
+ ('name:pool4,pool5', ['pool4']),
+])
+def test_filter_by_name(
+ verify_unpaged, pool_factory, input, expected_pool_names):
+ db.session.add(pool_factory(id=1, names=['pool1']))
+ db.session.add(pool_factory(id=2, names=['pool2']))
+ db.session.add(pool_factory(id=3, names=['pool3']))
+ db.session.add(pool_factory(id=4, names=['pool4', 'pool5', 'pool6']))
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('category:cat1', ['t1', 't2']),
+ ('category:cat2', ['t3']),
+ ('category:cat1,cat2', ['t1', 't2', 't3']),
+])
+def test_filter_by_category(
+ verify_unpaged,
+ pool_factory,
+ pool_category_factory,
+ input,
+ expected_pool_names):
+ cat1 = pool_category_factory(name='cat1')
+ cat2 = pool_category_factory(name='cat2')
+ pool1 = pool_factory(id=1, names=['t1'], category=cat1)
+ pool2 = pool_factory(id=2, names=['t2'], category=cat1)
+ pool3 = pool_factory(id=3, names=['t3'], category=cat2)
+ db.session.add_all([pool1, pool2, pool3])
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('creation-time:2014', ['t1', 't2']),
+ ('creation-date:2014', ['t1', 't2']),
+ ('-creation-time:2014', ['t3']),
+ ('-creation-date:2014', ['t3']),
+ ('creation-time:2014..2014-06', ['t1', 't2']),
+ ('creation-time:2014-06..2015-01-01', ['t2', 't3']),
+ ('creation-time:2014-06..', ['t2', 't3']),
+ ('creation-time:..2014-06', ['t1', 't2']),
+ ('-creation-time:2014..2014-06', ['t3']),
+ ('-creation-time:2014-06..2015-01-01', ['t1']),
+ ('creation-date:2014..2014-06', ['t1', 't2']),
+ ('creation-date:2014-06..2015-01-01', ['t2', 't3']),
+ ('creation-date:2014-06..', ['t2', 't3']),
+ ('creation-date:..2014-06', ['t1', 't2']),
+ ('-creation-date:2014..2014-06', ['t3']),
+ ('-creation-date:2014-06..2015-01-01', ['t1']),
+ ('creation-time:2014-01,2015', ['t1', 't3']),
+ ('creation-date:2014-01,2015', ['t1', 't3']),
+ ('-creation-time:2014-01,2015', ['t2']),
+ ('-creation-date:2014-01,2015', ['t2']),
+])
+def test_filter_by_creation_time(
+ verify_unpaged, pool_factory, input, expected_pool_names):
+ pool1 = pool_factory(id=1, names=['t1'])
+ pool2 = pool_factory(id=2, names=['t2'])
+ pool3 = pool_factory(id=3, names=['t3'])
+ pool1.creation_time = datetime(2014, 1, 1)
+ pool2.creation_time = datetime(2014, 6, 1)
+ pool3.creation_time = datetime(2015, 1, 1)
+ db.session.add_all([pool1, pool2, pool3])
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('last-edit-date:2014', ['t1', 't3']),
+ ('last-edit-time:2014', ['t1', 't3']),
+ ('edit-date:2014', ['t1', 't3']),
+ ('edit-time:2014', ['t1', 't3']),
+])
+def test_filter_by_edit_time(
+ verify_unpaged, pool_factory, input, expected_pool_names):
+ pool1 = pool_factory(id=1, names=['t1'])
+ pool2 = pool_factory(id=2, names=['t2'])
+ pool3 = pool_factory(id=3, names=['t3'])
+ pool1.last_edit_time = datetime(2014, 1, 1)
+ pool2.last_edit_time = datetime(2015, 1, 1)
+ pool3.last_edit_time = datetime(2014, 1, 1)
+ db.session.add_all([pool1, pool2, pool3])
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('post-count:2', ['t1']),
+ ('post-count:1', ['t2']),
+ ('post-count:1..', ['t1', 't2']),
+ ('post-count-min:1', ['t1', 't2']),
+ ('post-count:..1', ['t2']),
+ ('post-count-max:1', ['t2']),
+])
+def test_filter_by_post_count(
+ verify_unpaged, pool_factory, post_factory, input, expected_pool_names):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ pool1 = pool_factory(id=1, names=['t1'])
+ pool2 = pool_factory(id=2, names=['t2'])
+ db.session.add_all([post1, post2, pool1, pool2])
+ pool1.posts.append(post1)
+ pool1.posts.append(post2)
+ pool2.posts.append(post1)
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input', [
+ 'post-count:..',
+ 'post-count:asd',
+ 'post-count:asd,1',
+ 'post-count:1,asd',
+ 'post-count:asd..1',
+ 'post-count:1..asd',
+])
+def test_filter_by_invalid_input(executor, input):
+ with pytest.raises(errors.SearchError):
+ executor.execute(input, offset=0, limit=100)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('', ['t1', 't2']),
+ ('sort:name', ['t1', 't2']),
+ ('-sort:name', ['t2', 't1']),
+ ('sort:name,asc', ['t1', 't2']),
+ ('sort:name,desc', ['t2', 't1']),
+ ('-sort:name,asc', ['t2', 't1']),
+ ('-sort:name,desc', ['t1', 't2']),
+])
+def test_sort_by_name(verify_unpaged, pool_factory, input, expected_pool_names):
+ db.session.add(pool_factory(id=2, names=['t2']))
+ db.session.add(pool_factory(id=1, names=['t1']))
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('', ['t1', 't2', 't3']),
+ ('sort:creation-date', ['t3', 't2', 't1']),
+ ('sort:creation-time', ['t3', 't2', 't1']),
+])
+def test_sort_by_creation_time(
+ verify_unpaged, pool_factory, input, expected_pool_names):
+ pool1 = pool_factory(id=1, names=['t1'])
+ pool2 = pool_factory(id=2, names=['t2'])
+ pool3 = pool_factory(id=3, names=['t3'])
+ pool1.creation_time = datetime(1991, 1, 1)
+ pool2.creation_time = datetime(1991, 1, 2)
+ pool3.creation_time = datetime(1991, 1, 3)
+ db.session.add_all([pool3, pool1, pool2])
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('', ['t1', 't2', 't3']),
+ ('sort:last-edit-date', ['t3', 't2', 't1']),
+ ('sort:last-edit-time', ['t3', 't2', 't1']),
+ ('sort:edit-date', ['t3', 't2', 't1']),
+ ('sort:edit-time', ['t3', 't2', 't1']),
+])
+def test_sort_by_last_edit_time(
+ verify_unpaged, pool_factory, input, expected_pool_names):
+ pool1 = pool_factory(id=1, names=['t1'])
+ pool2 = pool_factory(id=2, names=['t2'])
+ pool3 = pool_factory(id=3, names=['t3'])
+ pool1.last_edit_time = datetime(1991, 1, 1)
+ pool2.last_edit_time = datetime(1991, 1, 2)
+ pool3.last_edit_time = datetime(1991, 1, 3)
+ db.session.add_all([pool3, pool1, pool2])
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('sort:post-count', ['t2', 't1']),
+])
+def test_sort_by_post_count(
+ verify_unpaged, pool_factory, post_factory, input, expected_pool_names):
+ post1 = post_factory(id=1)
+ post2 = post_factory(id=2)
+ pool1 = pool_factory(id=1, names=['t1'])
+ pool2 = pool_factory(id=2, names=['t2'])
+ db.session.add_all([post1, post2, pool1, pool2])
+ pool1.posts.append(post1)
+ pool2.posts.append(post1)
+ pool2.posts.append(post2)
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)
+
+
+@pytest.mark.parametrize('input,expected_pool_names', [
+ ('sort:category', ['t3', 't1', 't2']),
+])
+def test_sort_by_category(
+ verify_unpaged,
+ pool_factory,
+ pool_category_factory,
+ input,
+ expected_pool_names):
+ cat1 = pool_category_factory(name='cat1')
+ cat2 = pool_category_factory(name='cat2')
+ pool1 = pool_factory(id=1, names=['t1'], category=cat2)
+ pool2 = pool_factory(id=2, names=['t2'], category=cat2)
+ pool3 = pool_factory(id=3, names=['t3'], category=cat1)
+ db.session.add_all([pool1, pool2, pool3])
+ db.session.flush()
+ verify_unpaged(input, expected_pool_names)