diff options
Diffstat (limited to 'server/szurubooru/tests/func/test_posts.py')
| -rw-r--r-- | server/szurubooru/tests/func/test_posts.py | 938 |
1 files changed, 558 insertions, 380 deletions
diff --git a/server/szurubooru/tests/func/test_posts.py b/server/szurubooru/tests/func/test_posts.py index 70a521a..af6121a 100644 --- a/server/szurubooru/tests/func/test_posts.py +++ b/server/szurubooru/tests/func/test_posts.py @@ -1,40 +1,58 @@ +import os from datetime import datetime from unittest.mock import patch -import os + import pytest + from szurubooru import db, model from szurubooru.func import ( - posts, users, comments, tags, images, files, util, image_hash) + comments, + files, + image_hash, + images, + posts, + tags, + users, + util, +) -@pytest.mark.parametrize('input_mime_type,expected_url', [ - ('image/jpeg', 'http://example.com/posts/1_244c8840887984c4.jpg'), - ('image/gif', 'http://example.com/posts/1_244c8840887984c4.gif'), - ('totally/unknown', 'http://example.com/posts/1_244c8840887984c4.dat'), -]) +@pytest.mark.parametrize( + "input_mime_type,expected_url", + [ + ("image/jpeg", "http://example.com/posts/1_244c8840887984c4.jpg"), + ("image/gif", "http://example.com/posts/1_244c8840887984c4.gif"), + ("totally/unknown", "http://example.com/posts/1_244c8840887984c4.dat"), + ], +) def test_get_post_url(input_mime_type, expected_url, config_injector): - config_injector({'data_url': 'http://example.com/', 'secret': 'test'}) + config_injector({"data_url": "http://example.com/", "secret": "test"}) post = model.Post() post.post_id = 1 post.mime_type = input_mime_type assert posts.get_post_content_url(post) == expected_url -@pytest.mark.parametrize('input_mime_type', ['image/jpeg', 'image/gif']) +@pytest.mark.parametrize("input_mime_type", ["image/jpeg", "image/gif"]) def test_get_post_thumbnail_url(input_mime_type, config_injector): - config_injector({'data_url': 'http://example.com/', 'secret': 'test'}) + config_injector({"data_url": "http://example.com/", "secret": "test"}) post = model.Post() post.post_id = 1 post.mime_type = input_mime_type - assert posts.get_post_thumbnail_url(post) \ - == 'http://example.com/generated-thumbnails/1_244c8840887984c4.jpg' + assert ( + posts.get_post_thumbnail_url(post) + == "http://example.com/generated-thumbnails/1_244c8840887984c4.jpg" + ) -@pytest.mark.parametrize('input_mime_type,expected_path', [ - ('image/jpeg', 'posts/1_244c8840887984c4.jpg'), - ('image/gif', 'posts/1_244c8840887984c4.gif'), - ('totally/unknown', 'posts/1_244c8840887984c4.dat'), -]) +@pytest.mark.parametrize( + "input_mime_type,expected_path", + [ + ("image/jpeg", "posts/1_244c8840887984c4.jpg"), + ("image/gif", "posts/1_244c8840887984c4.gif"), + ("totally/unknown", "posts/1_244c8840887984c4.dat"), + ], +) def test_get_post_content_path(input_mime_type, expected_path): post = model.Post() post.post_id = 1 @@ -42,31 +60,35 @@ def test_get_post_content_path(input_mime_type, expected_path): assert posts.get_post_content_path(post) == expected_path -@pytest.mark.parametrize('input_mime_type', ['image/jpeg', 'image/gif']) +@pytest.mark.parametrize("input_mime_type", ["image/jpeg", "image/gif"]) def test_get_post_thumbnail_path(input_mime_type): post = model.Post() post.post_id = 1 post.mime_type = input_mime_type - assert posts.get_post_thumbnail_path(post) \ - == 'generated-thumbnails/1_244c8840887984c4.jpg' + assert ( + posts.get_post_thumbnail_path(post) + == "generated-thumbnails/1_244c8840887984c4.jpg" + ) -@pytest.mark.parametrize('input_mime_type', ['image/jpeg', 'image/gif']) +@pytest.mark.parametrize("input_mime_type", ["image/jpeg", "image/gif"]) def test_get_post_thumbnail_backup_path(input_mime_type): post = model.Post() post.post_id = 1 post.mime_type = input_mime_type - assert posts.get_post_thumbnail_backup_path(post) \ - == 'posts/custom-thumbnails/1_244c8840887984c4.dat' + assert ( + posts.get_post_thumbnail_backup_path(post) + == "posts/custom-thumbnails/1_244c8840887984c4.dat" + ) def test_serialize_note(): note = model.PostNote() note.polygon = [[0, 1], [1, 1], [1, 0], [0, 0]] - note.text = '...' + note.text = "..." assert posts.serialize_note(note) == { - 'polygon': [[0, 1], [1, 1], [1, 0], [0, 0]], - 'text': '...' + "polygon": [[0, 1], [1, 1], [1, 0], [0, 0]], + "text": "...", } @@ -75,174 +97,227 @@ def test_serialize_post_when_empty(): def test_serialize_post( - user_factory, - comment_factory, - tag_factory, - tag_category_factory, - metric_factory, - post_metric_factory, - post_metric_range_factory, - config_injector): - config_injector({'data_url': 'http://example.com/', 'secret': 'test'}) - with patch('szurubooru.func.comments.serialize_comment'), \ - patch('szurubooru.func.users.serialize_micro_user'), \ - patch('szurubooru.func.posts.files.has'): + user_factory, + comment_factory, + tag_factory, + tag_category_factory, + metric_factory, + post_metric_factory, + post_metric_range_factory, + pool_factory, + pool_category_factory, + config_injector, +): + config_injector({"data_url": "http://example.com/", "secret": "test"}) + with patch("szurubooru.func.comments.serialize_comment"), patch( + "szurubooru.func.users.serialize_micro_user" + ), patch("szurubooru.func.posts.files.has"): files.has.return_value = True - users.serialize_micro_user.side_effect \ - = lambda user, auth_user: user.name - comments.serialize_comment.side_effect \ - = lambda comment, auth_user: comment.user.name + users.serialize_micro_user.side_effect = ( + lambda user, auth_user: user.name + ) + comments.serialize_comment.side_effect = ( + lambda comment, auth_user: comment.user.name + ) - auth_user = user_factory(name='auth user') + auth_user = user_factory(name="auth user") post = model.Post() post.post_id = 1 post.creation_time = datetime(1997, 1, 1) post.last_edit_time = datetime(1998, 1, 1) tag1 = tag_factory( - names=['tag1', 'tag2'], - category=tag_category_factory('test-cat1')) + names=["tag1", "tag2"], + category=tag_category_factory("test-cat1") + ) tag1.metric = metric_factory(tag=tag1, min=-2.5, max=2.5) tag3 = tag_factory( - names=['tag3'], - category=tag_category_factory('test-cat2')) + names=["tag3"], + category=tag_category_factory("test-cat2") + ) post.tags = [tag1, tag3] post.metrics = [ post_metric_factory(post=post, metric=tag1.metric, value=-1.2) ] post.metric_ranges = [ - post_metric_range_factory(post=post, metric=tag1.metric, - low=2, high=3) + post_metric_range_factory( + post=post, metric=tag1.metric, low=2, high=3 + ) ] post.safety = model.Post.SAFETY_SAFE - post.source = '4gag' + post.source = "4gag" post.type = model.Post.TYPE_IMAGE - post.checksum = 'deadbeef' - post.mime_type = 'image/jpeg' + post.checksum = "deadbeef" + post.mime_type = "image/jpeg" post.file_size = 100 - post.user = user_factory(name='post author') + post.user = user_factory(name="post author") post.canvas_width = 200 post.canvas_height = 300 - post.flags = ['loop'] + post.flags = ["loop"] db.session.add(post) db.session.flush() - db.session.add_all([ - comment_factory( - user=user_factory(name='commenter1'), - post=post, - time=datetime(1999, 1, 1)), - comment_factory( - user=user_factory(name='commenter2'), - post=post, - time=datetime(1999, 1, 2)), - model.PostFavorite( - post=post, - user=user_factory(name='fav1'), - time=datetime(1800, 1, 1)), - model.PostFeature( - post=post, - user=user_factory(), - time=datetime(1999, 1, 1)), - model.PostScore( - post=post, - user=auth_user, - score=-1, - time=datetime(1800, 1, 1)), - model.PostScore( - post=post, - user=user_factory(), - score=1, - time=datetime(1800, 1, 1)), - model.PostScore( - post=post, - user=user_factory(), - score=1, - time=datetime(1800, 1, 1))]) + db.session.add_all( + [ + comment_factory( + user=user_factory(name="commenter1"), + post=post, + time=datetime(1999, 1, 1), + ), + comment_factory( + user=user_factory(name="commenter2"), + post=post, + time=datetime(1999, 1, 2), + ), + model.PostFavorite( + post=post, + user=user_factory(name="fav1"), + time=datetime(1800, 1, 1), + ), + model.PostFeature( + post=post, user=user_factory(), time=datetime(1999, 1, 1) + ), + model.PostScore( + post=post, + user=auth_user, + score=-1, + time=datetime(1800, 1, 1), + ), + model.PostScore( + post=post, + user=user_factory(), + score=1, + time=datetime(1800, 1, 1), + ), + model.PostScore( + post=post, + user=user_factory(), + score=1, + 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]) + result["tags"].sort(key=lambda tag: tag["names"][0]) assert result == { - 'id': 1, - 'version': 1, - 'creationTime': datetime(1997, 1, 1), - 'lastEditTime': datetime(1998, 1, 1), - 'safety': 'safe', - 'source': '4gag', - 'type': 'image', - 'checksum': 'deadbeef', - 'fileSize': 100, - 'canvasWidth': 200, - 'canvasHeight': 300, - 'contentUrl': 'http://example.com/posts/1_244c8840887984c4.jpg', - 'thumbnailUrl': - 'http://example.com/' - 'generated-thumbnails/1_244c8840887984c4.jpg', - 'flags': ['loop'], - 'tags': [ + "id": 1, + "version": 1, + "creationTime": datetime(1997, 1, 1), + "lastEditTime": datetime(1998, 1, 1), + "safety": "safe", + "source": "4gag", + "type": "image", + "checksum": "deadbeef", + "fileSize": 100, + "canvasWidth": 200, + "canvasHeight": 300, + "contentUrl": "http://example.com/posts/1_244c8840887984c4.jpg", + "thumbnailUrl": "http://example.com/" + "generated-thumbnails/1_244c8840887984c4.jpg", + "flags": ["loop"], + "tags": [ { - 'names': ['tag1', 'tag2'], - 'category': 'test-cat1', - 'usages': 1, - 'metric': { - 'min': -2.5, - 'max': 2.5 - } + "names": ["tag1", "tag2"], + "category": "test-cat1", + "usages": 1, + "metric": { + "min": -2.5, + "max": 2.5 + }, }, { - 'names': ['tag3'], - 'category': 'test-cat2', - 'usages': 1, - 'metric': None + "names": ["tag3"], + "category": "test-cat2", + "usages": 1, + "metric": None, }, ], - 'relations': [], - 'notes': [], - 'user': 'post author', - 'score': 1, - 'ownFavorite': False, - 'ownScore': -1, - 'tagCount': 2, - 'favoriteCount': 1, - 'commentCount': 2, - 'noteCount': 0, - 'featureCount': 1, - 'relationCount': 0, - 'lastFeatureTime': datetime(1999, 1, 1), - 'favoritedBy': ['fav1'], - 'hasCustomThumbnail': True, - 'mimeType': 'image/jpeg', - 'comments': ['commenter1', 'commenter2'], - 'metrics': [ + "relations": [], + "notes": [], + "pools": [ { - 'tag_name': 'tag1', - 'post_id': 1, - 'value': -1.2 + "id": 1, + "names": ["pool1", "pool2"], + "description": "desc", + "category": "test-cat1", + "postCount": 1, + }, + { + "id": 2, + "names": ["pool3"], + "description": "desc2", + "category": "test-cat2", + "postCount": 1, + }, + ], + "user": "post author", + "score": 1, + "ownFavorite": False, + "ownScore": -1, + "tagCount": 2, + "favoriteCount": 1, + "commentCount": 2, + "noteCount": 0, + "featureCount": 1, + "relationCount": 0, + "lastFeatureTime": datetime(1999, 1, 1), + "favoritedBy": ["fav1"], + "hasCustomThumbnail": True, + "mimeType": "image/jpeg", + "comments": ["commenter1", "commenter2"], + "metrics": [ + { + "tag_name": "tag1", + "post_id": 1, + "value": -1.2 } ], - 'metricRanges': [ + "metricRanges": [ { - 'tag_name': 'tag1', - 'post_id': 1, - 'low': 2, - 'high': 3 + "tag_name": "tag1", + "post_id": 1, + "low": 2, + "high": 3 } - ] + ], } def test_serialize_micro_post(post_factory, user_factory): - with patch('szurubooru.func.posts.get_post_thumbnail_url'): - posts.get_post_thumbnail_url.return_value \ - = 'https://example.com/thumb.png' + with patch("szurubooru.func.posts.get_post_thumbnail_url"): + posts.get_post_thumbnail_url.return_value = ( + "https://example.com/thumb.png" + ) auth_user = user_factory() post = post_factory() db.session.add(post) db.session.flush() assert posts.serialize_micro_post(post, auth_user) == { - 'id': post.post_id, - 'thumbnailUrl': 'https://example.com/thumb.png', + "id": post.post_id, + "thumbnailUrl": "https://example.com/thumb.png", } @@ -273,22 +348,25 @@ def test_get_post_by_id(post_factory): def test_create_post(user_factory, fake_datetime): - with patch('szurubooru.func.posts.update_post_content'), \ - patch('szurubooru.func.posts.update_post_tags'), \ - fake_datetime('1997-01-01'): + with patch("szurubooru.func.posts.update_post_content"), patch( + "szurubooru.func.posts.update_post_tags" + ), fake_datetime("1997-01-01"): auth_user = user_factory() - post, _new_tags = posts.create_post('content', ['tag'], auth_user) + post, _new_tags = posts.create_post("content", ["tag"], auth_user) assert post.creation_time == datetime(1997, 1, 1) assert post.last_edit_time is None - posts.update_post_tags.assert_called_once_with(post, ['tag']) - posts.update_post_content.assert_called_once_with(post, 'content') + posts.update_post_tags.assert_called_once_with(post, ["tag"]) + posts.update_post_content.assert_called_once_with(post, "content") -@pytest.mark.parametrize('input_safety,expected_safety', [ - ('safe', model.Post.SAFETY_SAFE), - ('sketchy', model.Post.SAFETY_SKETCHY), - ('unsafe', model.Post.SAFETY_UNSAFE), -]) +@pytest.mark.parametrize( + "input_safety,expected_safety", + [ + ("safe", model.Post.SAFETY_SAFE), + ("sketchy", model.Post.SAFETY_SKETCHY), + ("unsafe", model.Post.SAFETY_UNSAFE), + ], +) def test_update_post_safety(input_safety, expected_safety): post = model.Post() posts.update_post_safety(post, input_safety) @@ -298,95 +376,107 @@ def test_update_post_safety(input_safety, expected_safety): def test_update_post_safety_with_invalid_string(): post = model.Post() with pytest.raises(posts.InvalidPostSafetyError): - posts.update_post_safety(post, 'bad') + posts.update_post_safety(post, "bad") def test_update_post_source(): post = model.Post() - posts.update_post_source(post, 'x') - assert post.source == 'x' + posts.update_post_source(post, "x") + assert post.source == "x" def test_update_post_source_with_too_long_string(): post = model.Post() with pytest.raises(posts.InvalidPostSourceError): - posts.update_post_source(post, 'x' * 1000) + posts.update_post_source(post, "x" * 3000) @pytest.mark.parametrize( - 'is_existing,input_file,expected_mime_type,expected_type,output_file_name', + "is_existing,input_file,expected_mime_type,expected_type,output_file_name", [ ( True, - 'png.png', - 'image/png', + "png.png", + "image/png", model.Post.TYPE_IMAGE, - '1_244c8840887984c4.png', + "1_244c8840887984c4.png", ), ( False, - 'png.png', - 'image/png', + "png.png", + "image/png", model.Post.TYPE_IMAGE, - '1_244c8840887984c4.png', + "1_244c8840887984c4.png", ), ( False, - 'jpeg.jpg', - 'image/jpeg', + "jpeg.jpg", + "image/jpeg", model.Post.TYPE_IMAGE, - '1_244c8840887984c4.jpg', + "1_244c8840887984c4.jpg", ), ( False, - 'gif.gif', - 'image/gif', + "gif.gif", + "image/gif", model.Post.TYPE_IMAGE, - '1_244c8840887984c4.gif', + "1_244c8840887984c4.gif", ), ( False, - 'gif-animated.gif', - 'image/gif', + "gif-animated.gif", + "image/gif", model.Post.TYPE_ANIMATION, - '1_244c8840887984c4.gif', + "1_244c8840887984c4.gif", ), ( False, - 'webm.webm', - 'video/webm', + "webm.webm", + "video/webm", model.Post.TYPE_VIDEO, - '1_244c8840887984c4.webm', + "1_244c8840887984c4.webm", ), ( False, - 'mp4.mp4', - 'video/mp4', + "mp4.mp4", + "video/mp4", model.Post.TYPE_VIDEO, - '1_244c8840887984c4.mp4', + "1_244c8840887984c4.mp4", ), ( False, - 'flash.swf', - 'application/x-shockwave-flash', + "flash.swf", + "application/x-shockwave-flash", model.Post.TYPE_FLASH, - '1_244c8840887984c4.swf', + "1_244c8840887984c4.swf", ), - ]) + ], +) def test_update_post_content_for_new_post( - tmpdir, config_injector, post_factory, read_asset, is_existing, - input_file, expected_mime_type, expected_type, output_file_name): - with patch('szurubooru.func.util.get_sha1'): - util.get_sha1.return_value = 'crc' - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) - output_file_path = '{}/data/posts/{}'.format(tmpdir, output_file_name) + tmpdir, + config_injector, + post_factory, + read_asset, + is_existing, + input_file, + expected_mime_type, + expected_type, + output_file_name, +): + with patch("szurubooru.func.util.get_sha1"): + util.get_sha1.return_value = "crc" + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": False, + } + ) + output_file_path = "{}/data/posts/{}".format(tmpdir, output_file_name) post = post_factory(id=1) db.session.add(post) if is_existing: @@ -399,126 +489,157 @@ def test_update_post_content_for_new_post( db.session.flush() assert post.mime_type == expected_mime_type assert post.type == expected_type - assert post.checksum == 'crc' + assert post.checksum == "crc" assert os.path.exists(output_file_path) if post.type in (model.Post.TYPE_IMAGE, model.Post.TYPE_ANIMATION): - image_hash.delete_image.assert_called_once_with(post.post_id) - image_hash.add_image.assert_called_once_with(post.post_id, content) + assert db.session.query(model.PostSignature).count() == 1 else: - image_hash.delete_image.assert_not_called() - image_hash.add_image.assert_not_called() + assert db.session.query(model.PostSignature).count() == 0 def test_update_post_content_to_existing_content( - tmpdir, config_injector, post_factory, read_asset): - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'data_url': 'example.com', - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + tmpdir, config_injector, post_factory, read_asset +): + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "data_url": "example.com", + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": False, + } + ) post = post_factory() another_post = post_factory() db.session.add_all([post, another_post]) - posts.update_post_content(post, read_asset('png.png')) + posts.update_post_content(post, read_asset("png.png")) db.session.flush() with pytest.raises(posts.PostAlreadyUploadedError): - posts.update_post_content(another_post, read_asset('png.png')) + posts.update_post_content(another_post, read_asset("png.png")) +@pytest.mark.parametrize("allow_broken_uploads", [True, False]) def test_update_post_content_with_broken_content( - tmpdir, config_injector, post_factory, read_asset): + tmpdir, config_injector, post_factory, read_asset, allow_broken_uploads +): # the rationale behind this behavior is to salvage user upload even if the # server software thinks it's broken. chances are the server is wrong, # especially about flash movies. - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": allow_broken_uploads, + } + ) post = post_factory() another_post = post_factory() db.session.add_all([post, another_post]) - posts.update_post_content(post, read_asset('png-broken.png')) - db.session.flush() - assert post.canvas_width is None - assert post.canvas_height is None + if allow_broken_uploads: + posts.update_post_content(post, read_asset("png-broken.png")) + db.session.flush() + assert post.canvas_width is None + assert post.canvas_height is None + else: + with pytest.raises(posts.InvalidPostContentError): + posts.update_post_content(post, read_asset("png-broken.png")) + db.session.flush() -@pytest.mark.parametrize('input_content', [None, b'not a media file']) -def test_update_post_content_with_invalid_content(input_content): +@pytest.mark.parametrize("input_content", [None, b"not a media file"]) +def test_update_post_content_with_invalid_content( + config_injector, input_content +): + config_injector( + { + "allow_broken_uploads": True, + } + ) post = model.Post() with pytest.raises(posts.InvalidPostContentError): posts.update_post_content(post, input_content) -@pytest.mark.parametrize('is_existing', (True, False)) +@pytest.mark.parametrize("is_existing", (True, False)) def test_update_post_thumbnail_to_new_one( - tmpdir, config_injector, read_asset, post_factory, is_existing): - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + tmpdir, config_injector, read_asset, post_factory, is_existing +): + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": False, + } + ) post = post_factory(id=1) db.session.add(post) if is_existing: db.session.flush() assert post.post_id generated_path = ( - '{}/data/generated-thumbnails/1_244c8840887984c4.jpg' - .format(tmpdir)) + "{}/data/generated-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.jpg" + ) source_path = ( - '{}/data/posts/custom-thumbnails/1_244c8840887984c4.dat' - .format(tmpdir)) + "{}/data/posts/custom-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.dat" + ) assert not os.path.exists(generated_path) assert not os.path.exists(source_path) - posts.update_post_content(post, read_asset('png.png')) - posts.update_post_thumbnail(post, read_asset('jpeg.jpg')) + posts.update_post_content(post, read_asset("png.png")) + posts.update_post_thumbnail(post, read_asset("jpeg.jpg")) assert not os.path.exists(generated_path) assert not os.path.exists(source_path) db.session.flush() assert os.path.exists(generated_path) assert os.path.exists(source_path) - with open(source_path, 'rb') as handle: - assert handle.read() == read_asset('jpeg.jpg') + with open(source_path, "rb") as handle: + assert handle.read() == read_asset("jpeg.jpg") -@pytest.mark.parametrize('is_existing', (True, False)) +@pytest.mark.parametrize("is_existing", (True, False)) def test_update_post_thumbnail_to_default( - tmpdir, config_injector, read_asset, post_factory, is_existing): - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + tmpdir, config_injector, read_asset, post_factory, is_existing +): + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": False, + } + ) post = post_factory(id=1) db.session.add(post) if is_existing: db.session.flush() assert post.post_id generated_path = ( - '{}/data/generated-thumbnails/1_244c8840887984c4.jpg' - .format(tmpdir)) + "{}/data/generated-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.jpg" + ) source_path = ( - '{}/data/posts/custom-thumbnails/1_244c8840887984c4.dat' - .format(tmpdir)) + "{}/data/posts/custom-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.dat" + ) assert not os.path.exists(generated_path) assert not os.path.exists(source_path) - posts.update_post_content(post, read_asset('png.png')) - posts.update_post_thumbnail(post, read_asset('jpeg.jpg')) + posts.update_post_content(post, read_asset("png.png")) + posts.update_post_thumbnail(post, read_asset("jpeg.jpg")) posts.update_post_thumbnail(post, None) assert not os.path.exists(generated_path) assert not os.path.exists(source_path) @@ -527,80 +648,94 @@ def test_update_post_thumbnail_to_default( assert not os.path.exists(source_path) -@pytest.mark.parametrize('is_existing', (True, False)) +@pytest.mark.parametrize("is_existing", (True, False)) def test_update_post_thumbnail_with_broken_thumbnail( - tmpdir, config_injector, read_asset, post_factory, is_existing): - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + tmpdir, config_injector, read_asset, post_factory, is_existing +): + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": False, + } + ) post = post_factory(id=1) db.session.add(post) if is_existing: db.session.flush() assert post.post_id generated_path = ( - '{}/data/generated-thumbnails/1_244c8840887984c4.jpg' - .format(tmpdir)) + "{}/data/generated-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.jpg" + ) source_path = ( - '{}/data/posts/custom-thumbnails/1_244c8840887984c4.dat' - .format(tmpdir)) + "{}/data/posts/custom-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.dat" + ) assert not os.path.exists(generated_path) assert not os.path.exists(source_path) - posts.update_post_content(post, read_asset('png.png')) - posts.update_post_thumbnail(post, read_asset('png-broken.png')) + posts.update_post_content(post, read_asset("png.png")) + posts.update_post_thumbnail(post, read_asset("png-broken.png")) assert not os.path.exists(generated_path) assert not os.path.exists(source_path) db.session.flush() assert os.path.exists(generated_path) assert os.path.exists(source_path) - with open(source_path, 'rb') as handle: - assert handle.read() == read_asset('png-broken.png') - with open(generated_path, 'rb') as handle: + with open(source_path, "rb") as handle: + assert handle.read() == read_asset("png-broken.png") + with open(generated_path, "rb") as handle: image = images.Image(handle.read()) assert image.width == 1 assert image.height == 1 def test_update_post_content_leaving_custom_thumbnail( - tmpdir, config_injector, read_asset, post_factory): - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + tmpdir, config_injector, read_asset, post_factory +): + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + "allow_broken_uploads": False, + } + ) post = post_factory(id=1) db.session.add(post) - posts.update_post_content(post, read_asset('png.png')) - posts.update_post_thumbnail(post, read_asset('jpeg.jpg')) - posts.update_post_content(post, read_asset('png.png')) + posts.update_post_content(post, read_asset("png.png")) + posts.update_post_thumbnail(post, read_asset("jpeg.jpg")) + posts.update_post_content(post, read_asset("png.png")) db.session.flush() generated_path = ( - '{}/data/generated-thumbnails/1_244c8840887984c4.jpg' - .format(tmpdir)) + "{}/data/generated-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.jpg" + ) source_path = ( - '{}/data/posts/custom-thumbnails/1_244c8840887984c4.dat' - .format(tmpdir)) + "{}/data/posts/custom-thumbnails/".format(tmpdir) + + "1_244c8840887984c4.dat" + ) assert os.path.exists(source_path) assert os.path.exists(generated_path) def test_update_post_tags(tag_factory): post = model.Post() - with patch('szurubooru.func.tags.get_or_create_tags_by_names'): - tags.get_or_create_tags_by_names.side_effect = lambda tag_names: \ - ([tag_factory(names=[name]) for name in tag_names], []) - posts.update_post_tags(post, ['tag1', 'tag2']) + with patch("szurubooru.func.tags.get_or_create_tags_by_names"): + tags.get_or_create_tags_by_names.side_effect = lambda tag_names: ( + [tag_factory(names=[name]) for name in tag_names], + [], + ) + posts.update_post_tags(post, ["tag1", "tag2"]) assert len(post.tags) == 2 - assert post.tags[0].names[0].name == 'tag1' - assert post.tags[1].names[0].name == 'tag2' + assert post.tags[0].names[0].name == "tag1" + assert post.tags[1].names[0].name == "tag2" def test_update_post_relations(post_factory): @@ -612,7 +747,9 @@ def test_update_post_relations(post_factory): posts.update_post_relations(post, [relation1.post_id, relation2.post_id]) assert len(post.relations) == 2 assert sorted(r.post_id for r in post.relations) == [ - relation1.post_id, relation2.post_id] + relation1.post_id, + relation2.post_id, + ] def test_update_post_relations_bidirectionality(post_factory): @@ -647,35 +784,44 @@ def test_update_post_notes(): posts.update_post_notes( post, [ - {'polygon': [[0, 0], [0, 1], [1, 0], [0, 0]], 'text': 'text1'}, - {'polygon': [[0, 0], [0, 1], [1, 0], [0, 0]], 'text': 'text2'}, - ]) + {"polygon": [[0, 0], [0, 1], [1, 0], [0, 0]], "text": "text1"}, + {"polygon": [[0, 0], [0, 1], [1, 0], [0, 0]], "text": "text2"}, + ], + ) assert len(post.notes) == 2 assert post.notes[0].polygon == [[0, 0], [0, 1], [1, 0], [0, 0]] - assert post.notes[0].text == 'text1' + assert post.notes[0].text == "text1" assert post.notes[1].polygon == [[0, 0], [0, 1], [1, 0], [0, 0]] - assert post.notes[1].text == 'text2' + assert post.notes[1].text == "text2" -@pytest.mark.parametrize('input', [ - [{'text': '...'}], - [{'polygon': None, 'text': '...'}], - [{'polygon': 'trash', 'text': '...'}], - [{'polygon': ['trash', 'trash', 'trash'], 'text': '...'}], - [{'polygon': {2: 'trash', 3: 'trash', 4: 'trash'}, 'text': '...'}], - [{'polygon': [[0, 0]], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], None], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], 'surprise'], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], {2: 'trash', 3: 'trash'}], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], 5], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], [0, 2]], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], [0, '...']], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], [0, 0, 0]], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], [0]], 'text': '...'}], - [{'polygon': [[0, 0], [0, 0], [0, 1]], 'text': ''}], - [{'polygon': [[0, 0], [0, 0], [0, 1]], 'text': None}], - [{'polygon': [[0, 0], [0, 0], [0, 1]]}], -]) +@pytest.mark.parametrize( + "input", + [ + [{"text": "..."}], + [{"polygon": None, "text": "..."}], + [{"polygon": "trash", "text": "..."}], + [{"polygon": ["trash", "trash", "trash"], "text": "..."}], + [{"polygon": {2: "trash", 3: "trash", 4: "trash"}, "text": "..."}], + [{"polygon": [[0, 0]], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], None], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], "surprise"], "text": "..."}], + [ + { + "polygon": [[0, 0], [0, 0], {2: "trash", 3: "trash"}], + "text": "...", + } + ], + [{"polygon": [[0, 0], [0, 0], 5], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], [0, 2]], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], [0, "..."]], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], [0, 0, 0]], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], [0]], "text": "..."}], + [{"polygon": [[0, 0], [0, 0], [0, 1]], "text": ""}], + [{"polygon": [[0, 0], [0, 0], [0, 1]], "text": None}], + [{"polygon": [[0, 0], [0, 0], [0, 1]]}], + ], +) def test_update_post_notes_with_invalid_content(input): post = model.Post() with pytest.raises(posts.InvalidPostNoteError): @@ -684,14 +830,14 @@ def test_update_post_notes_with_invalid_content(input): def test_update_post_flags(): post = model.Post() - posts.update_post_flags(post, ['loop']) - assert post.flags == ['loop'] + posts.update_post_flags(post, ["loop"]) + assert post.flags == ["loop"] def test_update_post_flags_with_invalid_content(): post = model.Post() with pytest.raises(posts.InvalidPostFlagError): - posts.update_post_flags(post, ['invalid']) + posts.update_post_flags(post, ["invalid"]) def test_feature_post(post_factory, user_factory): @@ -707,7 +853,7 @@ def test_feature_post(post_factory, user_factory): def test_delete(post_factory, config_injector): - config_injector({'delete_source_files': False}) + config_injector({"delete_source_files": False}) post = post_factory() db.session.add(post) db.session.flush() @@ -718,7 +864,7 @@ def test_delete(post_factory, config_injector): def test_merge_posts_deletes_source_post(post_factory, config_injector): - config_injector({'delete_source_files': False}) + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() db.session.add_all([source_post, target_post]) @@ -731,7 +877,7 @@ def test_merge_posts_deletes_source_post(post_factory, config_injector): def test_merge_posts_with_itself(post_factory, config_injector): - config_injector({'delete_source_files': False}) + config_injector({"delete_source_files": False}) source_post = post_factory() db.session.add(source_post) db.session.flush() @@ -740,7 +886,7 @@ def test_merge_posts_with_itself(post_factory, config_injector): def test_merge_posts_moves_tags(post_factory, tag_factory, config_injector): - config_injector({'delete_source_files': False}) + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() tag = tag_factory() @@ -756,8 +902,9 @@ def test_merge_posts_moves_tags(post_factory, tag_factory, config_injector): def test_merge_posts_doesnt_duplicate_tags( - post_factory, tag_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, tag_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() tag = tag_factory() @@ -773,8 +920,9 @@ def test_merge_posts_doesnt_duplicate_tags( def test_merge_posts_moves_comments( - post_factory, comment_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, comment_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() comment = comment_factory(post=source_post) @@ -789,8 +937,9 @@ def test_merge_posts_moves_comments( def test_merge_posts_moves_scores( - post_factory, post_score_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, post_score_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() score = post_score_factory(post=source_post, score=1) @@ -805,8 +954,9 @@ def test_merge_posts_moves_scores( def test_merge_posts_doesnt_duplicate_scores( - post_factory, user_factory, post_score_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, user_factory, post_score_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() user = user_factory() @@ -823,8 +973,9 @@ def test_merge_posts_doesnt_duplicate_scores( def test_merge_posts_moves_favorites( - post_factory, post_favorite_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, post_favorite_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() favorite = post_favorite_factory(post=source_post) @@ -839,8 +990,9 @@ def test_merge_posts_moves_favorites( def test_merge_posts_doesnt_duplicate_favorites( - post_factory, user_factory, post_favorite_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, user_factory, post_favorite_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() user = user_factory() @@ -857,7 +1009,7 @@ def test_merge_posts_doesnt_duplicate_favorites( def test_merge_posts_moves_child_relations(post_factory, config_injector): - config_injector({'delete_source_files': False}) + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() related_post = post_factory() @@ -873,8 +1025,9 @@ def test_merge_posts_moves_child_relations(post_factory, config_injector): def test_merge_posts_doesnt_duplicate_child_relations( - post_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() related_post = post_factory() @@ -891,7 +1044,7 @@ def test_merge_posts_doesnt_duplicate_child_relations( def test_merge_posts_moves_parent_relations(post_factory, config_injector): - config_injector({'delete_source_files': False}) + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() related_post = post_factory() @@ -909,8 +1062,9 @@ def test_merge_posts_moves_parent_relations(post_factory, config_injector): def test_merge_posts_doesnt_duplicate_parent_relations( - post_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() related_post = post_factory() @@ -928,8 +1082,9 @@ def test_merge_posts_doesnt_duplicate_parent_relations( def test_merge_posts_doesnt_create_relation_loop_for_children( - post_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() source_post.relations = [target_post] @@ -944,8 +1099,9 @@ def test_merge_posts_doesnt_create_relation_loop_for_children( def test_merge_posts_doesnt_create_relation_loop_for_parents( - post_factory, config_injector): - config_injector({'delete_source_files': False}) + post_factory, config_injector +): + config_injector({"delete_source_files": False}) source_post = post_factory() target_post = post_factory() target_post.relations = [source_post] @@ -960,30 +1116,36 @@ def test_merge_posts_doesnt_create_relation_loop_for_parents( def test_merge_posts_replaces_content( - post_factory, config_injector, tmpdir, read_asset): - config_injector({ - 'data_dir': str(tmpdir.mkdir('data')), - 'data_url': 'example.com', - 'delete_source_files': False, - 'thumbnails': { - 'post_width': 300, - 'post_height': 300, - }, - 'secret': 'test', - }) + post_factory, config_injector, tmpdir, read_asset +): + config_injector( + { + "data_dir": str(tmpdir.mkdir("data")), + "data_url": "example.com", + "delete_source_files": False, + "thumbnails": { + "post_width": 300, + "post_height": 300, + }, + "secret": "test", + } + ) source_post = post_factory(id=1) target_post = post_factory(id=2) - content = read_asset('png.png') + content = read_asset("png.png") db.session.add_all([source_post, target_post]) db.session.commit() posts.update_post_content(source_post, content) db.session.flush() - source_path = ( - os.path.join('{}/data/posts/1_244c8840887984c4.png'.format(tmpdir))) - target_path1 = ( - os.path.join('{}/data/posts/2_49caeb3ec1643406.png'.format(tmpdir))) - target_path2 = ( - os.path.join('{}/data/posts/2_49caeb3ec1643406.dat'.format(tmpdir))) + source_path = os.path.join( + "{}/data/posts/1_244c8840887984c4.png".format(tmpdir) + ) + target_path1 = os.path.join( + "{}/data/posts/2_49caeb3ec1643406.png".format(tmpdir) + ) + target_path2 = os.path.join( + "{}/data/posts/2_49caeb3ec1643406.dat".format(tmpdir) + ) assert os.path.exists(source_path) assert not os.path.exists(target_path1) assert not os.path.exists(target_path2) @@ -995,3 +1157,19 @@ def test_merge_posts_replaces_content( assert os.path.exists(source_path) assert os.path.exists(target_path1) assert not os.path.exists(target_path2) + + +def test_search_by_image(post_factory, config_injector, read_asset): + config_injector({"allow_broken_uploads": False}) + post = post_factory() + posts.generate_post_signature(post, read_asset("jpeg.jpg")) + db.session.flush() + + result1 = posts.search_by_image(read_asset("jpeg-similar.jpg")) + assert len(result1) == 1 + result1_distance, result1_post = result1[0] + assert abs(result1_distance - 0.19713075553164386) < 1e-8 + assert result1_post.post_id == post.post_id + + result2 = posts.search_by_image(read_asset("png.png")) + assert not result2 |