Add pool tests

This commit is contained in:
Ruin0x11 2020-05-04 19:12:54 -07:00
parent 6b8e3f251f
commit 5ca21f9e7f
18 changed files with 1292 additions and 7 deletions

View file

@ -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

View file

@ -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()

View file

@ -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)

View file

@ -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')

View file

@ -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')

View file

@ -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)))

View file

@ -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

View file

@ -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': '-'})

View file

@ -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)

View file

@ -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)))

View file

@ -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

View file

@ -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)))

View file

@ -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})

View file

@ -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})

View file

@ -201,6 +201,52 @@ def post_favorite_factory(user_factory, post_factory):
return 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):

View file

@ -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,

View file

@ -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

View file

@ -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)