From c56b2decb17fc42261d8bc2fed00980725474714 Mon Sep 17 00:00:00 2001 From: Qynn Swaan Date: Sun, 12 Feb 2023 17:38:18 -0500 Subject: [PATCH] [#676] [bukuserver API] improve api views --- buku | 28 ++-- bukuserver/api.py | 149 ++++++++++--------- bukuserver/forms.py | 87 +++++++++-- bukuserver/response.py | 52 ++++++- bukuserver/server.py | 31 ++-- bukuserver/views.py | 5 +- tests/test_bukuDb.py | 2 +- tests/test_server.py | 319 ++++++++++++++++++++++++----------------- 8 files changed, 424 insertions(+), 249 deletions(-) diff --git a/buku b/buku index 97fd6c6e..121b7781 100755 --- a/buku +++ b/buku @@ -1959,7 +1959,7 @@ class BukuDb: return parse_tags(tags) - def replace_tag(self, orig: str, new: List[str] = []) -> bool: + def replace_tag(self, orig: str, new: List[str] = []): """Replace original tag by new tags in all records. Remove original tag if new tag is empty. @@ -1971,22 +1971,26 @@ class BukuDb: new : list Replacement tags. - Returns + Raises ------- - bool - True on success, False on failure. + ValueError: Invalid input(s) provided. + RuntimeError: Tag deletion failed. + """ + if DELIM in orig: + raise ValueError("Original tag cannot contain delimiter ({}).".format(DELIM)) + orig = delim_wrap(orig) - newtags = parse_tags(new) if new else DELIM + newtags = parse_tags([DELIM.join(new)]) if orig == newtags: - print('Tags are same.') - return False + raise ValueError("Original and replacement tags are the same.") # Remove original tag from DB if new tagset reduces to delimiter if newtags == DELIM: - return self.delete_tag_at_index(0, orig) + if not self.delete_tag_at_index(0, orig): + raise RuntimeError("Tag deletion failed.") # Update bookmarks with original tag query = 'SELECT id, tags FROM bookmarks WHERE tags LIKE ?' @@ -2002,8 +2006,6 @@ class BukuDb: self.conn.commit() - return True - def get_tagstr_from_taglist(self, id_list, taglist): """Get a string of delimiter-separated (and enclosed) string of tags from a dictionary of tags by matching ids. @@ -5974,7 +5976,11 @@ POSITIONAL ARGUMENTS: if len(args.replace) == 1: bdb.delete_tag_at_index(0, args.replace[0]) else: - bdb.replace_tag(args.replace[0], args.replace[1:]) + try: + bdb.replace_tag(args.replace[0], [' '.join(args.replace[1:])]) + except Exception as e: + LOGERR(str(e)) + bdb.close_quit(1) # Export bookmarks if args.export is not None and not search_opted: diff --git a/bukuserver/api.py b/bukuserver/api.py index 171ffb8a..8b34412d 100644 --- a/bukuserver/api.py +++ b/bukuserver/api.py @@ -6,29 +6,23 @@ from unittest import mock from flask.views import MethodView -from flask_api import exceptions, status import buku from buku import BukuDb import flask -from flask import current_app, jsonify, redirect, request, url_for +from flask import current_app, redirect, request, url_for try: - from . import forms, response + from response import Response + from forms import ApiBookmarkCreateForm, ApiBookmarkEditForm, ApiBookmarkRangeEditForm, ApiTagForm except ImportError: - from bukuserver import forms, response + from bukuserver.response import Response + from bukuserver.forms import ApiBookmarkCreateForm, ApiBookmarkEditForm, ApiBookmarkRangeEditForm, ApiTagForm STATISTIC_DATA = None -response_ok = lambda: (jsonify(response.response_template['success']), - status.HTTP_200_OK, - {'ContentType': 'application/json'}) -response_bad = lambda: (jsonify(response.response_template['failure']), - status.HTTP_400_BAD_REQUEST, - {'ContentType': 'application/json'}) -to_response = lambda ok: response_ok() if ok else response_bad() def entity(bookmark, id=False): data = { @@ -94,23 +88,35 @@ class ApiTagView(MethodView): def get(self, tag: T.Optional[str]): bukudb = get_bukudb() if tag is None: - return {"tags": search_tag(db=bukudb, limit=5)[0]} + return Response.SUCCESS(data={"tags": search_tag(db=bukudb, limit=5)[0]}) tags = search_tag(db=bukudb, stag=tag) if tag not in tags[1]: - raise exceptions.NotFound() - return {"name": tag, "usage_count": tags[1][tag]} + return Response.TAG_NOT_FOUND() + return Response.SUCCESS(data={"name": tag, "usage_count": tags[1][tag]}) def put(self, tag: str): + form = ApiTagForm({}) + error_response, data = form.process_data(request.get_json()) + if error_response is not None: + return error_response(data=data) bukudb = get_bukudb() + tags = search_tag(db=bukudb, stag=tag) + if tag not in tags[1]: + return Response.TAG_NOT_FOUND() try: - new_tags = request.data.get('tags') # type: ignore - if new_tags: - new_tags = new_tags.split(',') - else: - return response_bad() - except AttributeError as e: - raise exceptions.ParseError(detail=str(e)) - return to_response(bukudb.replace_tag(tag, new_tags)) + bukudb.replace_tag(tag, form.tags.data) + return Response.SUCCESS() + except (ValueError, RuntimeError): + return Response.FAILURE() + + def delete(self, tag: str): + if buku.DELIM in tag: + return Response.TAG_NOT_VALID() + bukudb = get_bukudb() + tags = search_tag(db=bukudb, stag=tag) + if tag not in tags[1]: + return Response.TAG_NOT_FOUND() + return Response.from_flag(bukudb.delete_tag_at_index(0, tag, chatty=False)) class ApiBookmarkView(MethodView): @@ -121,34 +127,40 @@ def get(self, rec_id: T.Union[int, None]): all_bookmarks = bukudb.get_rec_all() result = {'bookmarks': [entity(bookmark, id=not request.path.startswith('/api/')) for bookmark in all_bookmarks]} - res = jsonify(result) else: bukudb = getattr(flask.g, 'bukudb', get_bukudb()) bookmark = bukudb.get_rec_by_id(rec_id) - res = (response_bad() if bookmark is None else jsonify(entity(bookmark))) - return res + if bookmark is None: + return Response.BOOKMARK_NOT_FOUND() + result = entity(bookmark) + return Response.SUCCESS(data=result) def post(self, rec_id: None = None): + form = ApiBookmarkCreateForm({}) + error_response, error_data = form.process_data(request.get_json()) + if error_response is not None: + return error_response(data=error_data) bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - create_bookmarks_form = forms.ApiBookmarkForm() - url_data = create_bookmarks_form.url.data result_flag = bukudb.add_rec( - url_data, - create_bookmarks_form.title.data, - create_bookmarks_form.tags.data, - create_bookmarks_form.description.data - ) - return to_response(result_flag) + form.url.data, + form.title.data, + form.tags_str, + form.description.data) + return Response.from_flag(result_flag) def put(self, rec_id: int): + form = ApiBookmarkEditForm({}) + error_response, error_data = form.process_data(request.get_json()) + if error_response is not None: + return error_response(data=error_data) bukudb = getattr(flask.g, 'bukudb', get_bukudb()) result_flag = bukudb.update_rec( rec_id, - request.form.get('url'), - request.form.get('title'), - request.form.get('tags'), - request.form.get('description')) - return to_response(result_flag) + form.url.data, + form.title.data, + form.tags_str, + form.description.data) + return Response.from_flag(result_flag) def delete(self, rec_id: T.Union[int, None]): if rec_id is None: @@ -158,7 +170,7 @@ def delete(self, rec_id: T.Union[int, None]): else: bukudb = getattr(flask.g, 'bukudb', get_bukudb()) result_flag = bukudb.delete_rec(rec_id) - return to_response(result_flag) + return Response.from_flag(result_flag) class ApiBookmarkRangeView(MethodView): @@ -166,37 +178,49 @@ class ApiBookmarkRangeView(MethodView): def get(self, starting_id: int, ending_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) max_id = bukudb.get_max_id() or 0 - if starting_id > max_id or ending_id > max_id: - return response_bad() + if starting_id > ending_id or ending_id > max_id: + return Response.RANGE_NOT_VALID() result = {'bookmarks': {i: entity(bukudb.get_rec_by_id(i)) for i in range(starting_id, ending_id + 1)}} - return jsonify(result) + return Response.SUCCESS(data=result) def put(self, starting_id: int, ending_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) max_id = bukudb.get_max_id() or 0 - if starting_id > max_id or ending_id > max_id: - return response_bad() - for i in range(starting_id, ending_id + 1, 1): - updated_bookmark = request.data.get(str(i)) # type: ignore - result_flag = bukudb.update_rec( - i, - updated_bookmark.get('url'), - updated_bookmark.get('title'), - updated_bookmark.get('tags'), - updated_bookmark.get('description')) - if result_flag is False: - return response_bad() - return response_ok() + if starting_id > ending_id or ending_id > max_id: + return Response.RANGE_NOT_VALID() + updates = [] + errors = {} + for rec_id in range(starting_id, ending_id + 1): + json = request.get_json().get(str(rec_id)) + if json is None: + errors[rec_id] = 'Input required.' + continue + form = ApiBookmarkRangeEditForm({}) + error_response, error_data = form.process_data(json) + if error_response is not None: + errors[rec_id] = error_data.get('errors') + updates += [{'index': rec_id, + 'url': form.url.data, + 'title_in': form.title.data, + 'tags_in': form.tags_in, + 'desc': form.description.data}] + + if errors: + return Response.INPUT_NOT_VALID(data={'errors': errors}) + for update in updates: + if not bukudb.update_rec(**update): + return Response.FAILURE() + return Response.SUCCESS() def delete(self, starting_id: int, ending_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) max_id = bukudb.get_max_id() or 0 - if starting_id > max_id or ending_id > max_id: - return response_bad() + if starting_id > ending_id or ending_id > max_id: + return Response.RANGE_NOT_VALID() idx = min([starting_id, ending_id]) result_flag = bukudb.delete_rec(idx, starting_id, ending_id, is_range=True) - return to_response(result_flag) + return Response.from_flag(result_flag) class ApiBookmarkSearchView(MethodView): @@ -217,14 +241,11 @@ def get(self): ) deep = deep if isinstance(deep, bool) else deep.lower() == 'true' regex = regex if isinstance(regex, bool) else regex.lower() == 'true' - bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - res = None result = {'bookmarks': [entity(bookmark, id=True) for bookmark in bukudb.searchdb(keywords, all_keywords, deep, regex)]} current_app.logger.debug('total bookmarks:{}'.format(len(result['bookmarks']))) - res = jsonify(result) - return res + return Response.SUCCESS(data=result) def delete(self): arg_obj = request.form @@ -246,8 +267,8 @@ def delete(self): res = None for bookmark in bukudb.searchdb(keywords, all_keywords, deep, regex): if not bukudb.delete_rec(bookmark.id): - res = response_bad() - return res or response_ok() + res = Response.FAILURE() + return res or Response.SUCCESS() class BookmarkletView(MethodView): # pylint: disable=too-few-public-methods diff --git a/bukuserver/forms.py b/bukuserver/forms.py index 5f328f22..b489cef4 100644 --- a/bukuserver/forms.py +++ b/bukuserver/forms.py @@ -1,26 +1,85 @@ """Forms module.""" # pylint: disable=too-few-public-methods, missing-docstring -import wtforms +from typing import Any, Dict, Tuple from flask_wtf import FlaskForm +from wtforms.fields import BooleanField, FieldList, StringField, TextAreaField, HiddenField +from wtforms.validators import DataRequired, InputRequired, ValidationError +from buku import DELIM, parse_tags +from bukuserver.response import Response + +def validate_tag(form, field): + if not isinstance(field.data, str): + raise ValidationError('Tag must be a string.') + if DELIM in field.data: + raise ValidationError('Tag must not contain delimiter \"{}\".'.format(DELIM)) class SearchBookmarksForm(FlaskForm): - keywords = wtforms.FieldList(wtforms.StringField('Keywords'), min_entries=1) - all_keywords = wtforms.BooleanField('Match all keywords') - deep = wtforms.BooleanField('Deep search') - regex = wtforms.BooleanField('Regex') + keywords = FieldList(StringField('Keywords'), min_entries=1) + all_keywords = BooleanField('Match all keywords') + deep = BooleanField('Deep search') + regex = BooleanField('Regex') class HomeForm(SearchBookmarksForm): - keyword = wtforms.StringField('Keyword') + keyword = StringField('Keyword') class BookmarkForm(FlaskForm): - url = wtforms.StringField('Url', name='link', validators=[wtforms.validators.InputRequired()]) - title = wtforms.StringField() - tags = wtforms.StringField() - description = wtforms.TextAreaField() - fetch = wtforms.HiddenField(filters=[bool]) - -class ApiBookmarkForm(BookmarkForm): - url = wtforms.StringField(validators=[wtforms.validators.DataRequired()]) + url = StringField('Url', name='link', validators=[InputRequired()]) + title = StringField() + tags = StringField() + description = TextAreaField() + fetch = HiddenField(filters=[bool]) + + +class ApiTagForm(FlaskForm): + class Meta: + csrf = False + + tags = FieldList(StringField(validators=[DataRequired(), validate_tag]), min_entries=1) + + tags_str = None + + def process_data(self, data: Dict[str, Any]) -> Tuple[Response, Dict[str, Any]]: + """Generate comma-separated string tags_str based on list of tags.""" + tags = data.get('tags') + if tags and not isinstance(tags, list): + return Response.INPUT_NOT_VALID, {'errors': {'tags': 'List of tags expected.'}} + + super().process(data=data) + if not self.validate(): + return Response.INPUT_NOT_VALID, {'errors': self.errors} + + self.tags_str = None if tags is None else parse_tags([DELIM.join(tags)]) + return None, None + + +class ApiBookmarkCreateForm(ApiTagForm): + class Meta: + csrf = False + + url = StringField(validators=[DataRequired()]) + title = StringField() + description = StringField() + tags = FieldList(StringField(validators=[validate_tag]), min_entries=0) + + +class ApiBookmarkEditForm(ApiBookmarkCreateForm): + url = StringField() + + +class ApiBookmarkRangeEditForm(ApiBookmarkEditForm): + + del_tags = BooleanField('Delete tags list from existing tags', default=False) + + tags_in = None + + def process_data(self, data: Dict[str, Any]) -> Tuple[Response, Dict[str, Any]]: + """Generate comma-separated string tags_in based on list of tags.""" + error_response, data = super().process_data(data) + + if self.tags_str is not None: + self.tags_in = ("-" if self.del_tags.data else "+") + self.tags_str + + return error_response, data diff --git a/bukuserver/response.py b/bukuserver/response.py index 69b83567..c66ba301 100644 --- a/bukuserver/response.py +++ b/bukuserver/response.py @@ -1,4 +1,48 @@ -response_template = { - "success": {'status': 0, 'message': 'success'}, - "failure": {'status': 1, 'message': 'failure'} -} +from typing import Any, Dict +from enum import Enum +from flask import jsonify +from flask_api.status import HTTP_200_OK, HTTP_400_BAD_REQUEST, HTTP_404_NOT_FOUND + +OK, FAIL = 0, 1 + + +class Response(Enum): + SUCCESS = (HTTP_200_OK, "Success.") + FAILURE = (HTTP_400_BAD_REQUEST, "Failure.") + INPUT_NOT_VALID = (HTTP_400_BAD_REQUEST, "Input data not valid.") + BOOKMARK_NOT_FOUND = (HTTP_404_NOT_FOUND, "Bookmark not found.") + TAG_NOT_FOUND = (HTTP_404_NOT_FOUND, "Tag not found.") + RANGE_NOT_VALID = (HTTP_400_BAD_REQUEST, "Range not valid.") + TAG_NOT_VALID = (HTTP_400_BAD_REQUEST, "Invalid tag.") + + @staticmethod + def bad_request(message: str): + json = {'status': Response.FAILURE.status, 'message': message} + return (jsonify(json), Response.FAILURE.status_code, {'ContentType': 'application/json'}) + + @staticmethod + def from_flag(flag: bool): + return Response.SUCCESS() if flag else Response.FAILURE() + + @property + def status_code(self) -> int: + return self.value[0] + + @property + def message(self) -> str: + return self.value[1] + + @property + def status(self) -> int: + return OK if self.status_code == HTTP_200_OK else FAIL + + def json(self, data: Dict[str, Any] = None) -> Dict[str, Any]: + return dict(status=self.status, message=self.message, **data or {}) # pylint: disable=R1735 + + def __call__(self, *, data: Dict[str, Any] = None): + """Generates a tuple in the form (response, status, headers) + + If passed, data is added to the response's JSON. + """ + + return (jsonify(self.json(data)), self.status_code, {'ContentType': 'application/json'}) diff --git a/bukuserver/server.py b/bukuserver/server.py index 2ec33a4c..7978a78b 100644 --- a/bukuserver/server.py +++ b/bukuserver/server.py @@ -8,7 +8,7 @@ from flask.cli import FlaskGroup from flask_admin import Admin -from flask_api import FlaskAPI, status +from flask_api import FlaskAPI from flask_bootstrap import Bootstrap from buku import BukuDb, __version__, network_handler @@ -20,45 +20,40 @@ import click import flask from flask import __version__ as flask_version # type: ignore -from flask import ( - current_app, - jsonify, - redirect, - request, - url_for, -) +from flask import current_app, redirect, request, url_for try: - from . import api, response, views + from . import api, views + from response import Response except ImportError: - from bukuserver import api, response, views + from bukuserver import api, views + from bukuserver.response import Response STATISTIC_DATA = None def handle_network(): - failed_resp = response.response_template['failure'], status.HTTP_400_BAD_REQUEST url = request.data.get('url', None) if not url: - return failed_resp + return Response.FAILURE() try: res = network_handler(url) keys = ['title', 'description', 'tags', 'recognized mime', 'bad url'] res_dict = dict(zip(keys, res)) - return jsonify(res_dict) + return Response.SUCCESS(data=res_dict) except Exception as e: current_app.logger.debug(str(e)) - return failed_resp + return Response.FAILURE() def refresh_bookmark(rec_id: Union[int, None]): result_flag = getattr(flask.g, 'bukudb', api.get_bukudb()).refreshdb(rec_id or 0, request.form.get('threads', 4)) - return api.to_response(result_flag) + return Response.from_flag(result_flag) def get_tiny_url(rec_id): url = getattr(flask.g, 'bukudb', api.get_bukudb()).tnyfy_url(rec_id) - return jsonify({'url': url}) if url else api.response_bad() + return Response.SUCCESS(data={'url': url}) if url else Response.FAILURE() _BOOL_VALUES = {'true': True, '1': True, 'false': False, '0': False} @@ -132,8 +127,8 @@ def shell_context(): # routing # api tag_api_view = api.ApiTagView.as_view('tag_api') - app.add_url_rule('/api/tags', defaults={'tag': None}, view_func=tag_api_view, methods=['GET']) - app.add_url_rule('/api/tags/', view_func=tag_api_view, methods=['GET', 'PUT']) + app.add_url_rule('/api/tags', defaults={'tag': None}, view_func=tag_api_view, methods=['GET'], strict_slashes=False) + app.add_url_rule('/api/tags/', view_func=tag_api_view, methods=['GET', 'PUT', 'DELETE']) bookmark_api_view = api.ApiBookmarkView.as_view('bookmark_api') app.add_url_rule('/api/bookmarks', defaults={'rec_id': None}, view_func=bookmark_api_view, methods=['GET', 'POST', 'DELETE']) app.add_url_rule('/api/bookmarks/', view_func=bookmark_api_view, methods=['GET', 'PUT', 'DELETE']) diff --git a/bukuserver/views.py b/bukuserver/views.py index 91c9ecad..8f46da17 100644 --- a/bukuserver/views.py +++ b/bukuserver/views.py @@ -566,12 +566,11 @@ def delete_model(self, model): return res def update_model(self, form, model): - res = None try: original_name = model.name form.populate_obj(model) self._on_model_change(form, model, False) - res = self.bukudb.replace_tag(original_name, [model.name]) + self.bukudb.replace_tag(original_name, [model.name]) self.all_tags = self.bukudb.get_tag_all() except Exception as ex: if not self.handle_view_exception(ex): @@ -583,7 +582,7 @@ def update_model(self, form, model): LOG.exception(msg) return False self.after_model_change(form, model, False) - return res + return True def create_model(self, form): pass diff --git a/tests/test_bukuDb.py b/tests/test_bukuDb.py index 7976990b..7066308d 100644 --- a/tests/test_bukuDb.py +++ b/tests/test_bukuDb.py @@ -995,7 +995,7 @@ def test_delete_rec_index_and_delay_commit(setup, index, delay_commit, input_ret elif n_index > db_len: assert not res assert len(bdb.get_rec_all()) == db_len - elif index == 0 and input_retval != "y": + elif index == 0 and not input_retval: assert not res assert len(bdb.get_rec_all()) == db_len else: diff --git a/tests/test_server.py b/tests/test_server.py index 90a157e8..115ece66 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,14 +1,29 @@ -import json - +from typing import Any, Dict import pytest import flask +from flask_api.status import HTTP_405_METHOD_NOT_ALLOWED from click.testing import CliRunner - from bukuserver import server -from bukuserver.response import response_template +from bukuserver.response import Response from bukuserver.server import get_bool_from_env_var +def assert_response(response, exp_res: Response, data: Dict[str, Any] = None): + assert response.status_code == exp_res.status_code + assert response.get_json() == exp_res.json(data=data) + + +@pytest.mark.parametrize( + 'data, exp_json', [ + [None, {'status': 0, 'message': 'Success.'}], + [{}, {'status': 0, 'message': 'Success.'}], + [{'key': 'value'}, {'status': 0, 'message': 'Success.', 'key': 'value'}], + ] +) +def test_response_json(data, exp_json): + assert Response.SUCCESS.json(data=data) == exp_json + + @pytest.mark.parametrize( 'args,word', [ @@ -38,204 +53,240 @@ def test_home(client): @pytest.mark.parametrize( - 'url, exp_res', [ - ['/api/tags', {'tags': []}], - ['/api/bookmarks', {'bookmarks': []}], - ['/api/bookmarks/search', {'bookmarks': []}], - ['/api/bookmarks/refresh', response_template['failure']] + 'method, url, exp_res, data', [ + ['get', '/api/tags', Response.SUCCESS, {'tags': []}], + ['get', '/api/bookmarks', Response.SUCCESS, {'bookmarks': []}], + ['get', '/api/bookmarks/search', Response.SUCCESS, {'bookmarks': []}], + ['post', '/api/bookmarks/refresh', Response.FAILURE, None] ] ) -def test_api_empty_db(client, url, exp_res): - if url == '/api/bookmarks/refresh': - rd = client.post(url) - assert rd.status_code == 400 - else: - rd = client.get(url) - assert rd.status_code == 200 - assert rd.get_json() == exp_res +def test_api_empty_db(client, method, url, exp_res, data): + rd = getattr(client, method)(url) + assert_response(rd, exp_res, data) @pytest.mark.parametrize( - 'url, exp_res, status_code, method', [ - ['/api/tags/1', {'message': 'This resource does not exist.'}, 404, 'get'], - ['/api/tags/1', response_template['failure'], 400, 'put'], - ['/api/bookmarks/1', response_template['failure'], 400, 'get'], - ['/api/bookmarks/1', response_template['failure'], 400, 'put'], - ['/api/bookmarks/1', response_template['failure'], 400, 'delete'], - ['/api/bookmarks/1/refresh', response_template['failure'], 400, 'post'], - ['/api/bookmarks/1/tiny', response_template['failure'], 400, 'get'], - ['/api/bookmarks/1/2', response_template['failure'], 400, 'get'], - ['/api/bookmarks/1/2', response_template['failure'], 400, 'put'], - ['/api/bookmarks/1/2', response_template['failure'], 400, 'delete'], + 'url, methods', [ + ['api/tags', ['post', 'put', 'delete']], + ['/api/tags/tag1', ['post']], + ['api/bookmarks', ['put']], + ['/api/bookmarks/1', ['post']], + ['/api/bookmarks/refresh', ['get', 'put', 'delete']], + ['api/bookmarks/1/refresh', ['get', 'put', 'delete']], + ['api/bookmarks/1/tiny', ['post', 'put', 'delete']], + ['/api/bookmarks/1/2', ['post']], ] ) -def test_invalid_id(client, url, exp_res, status_code, method): - rd = getattr(client, method)(url) - assert rd.status_code == status_code - assert rd.get_json() == exp_res +def test_not_allowed(client, url, methods): + for method in methods: + rd = getattr(client, method)(url) + assert rd.status_code == HTTP_405_METHOD_NOT_ALLOWED + + +@pytest.mark.parametrize( + 'method, url, json, exp_res', [ + ['get', '/api/tags/tag1', None, Response.TAG_NOT_FOUND], + ['put', '/api/tags/tag1', {'tags': ['tag2']}, Response.TAG_NOT_FOUND], + ['delete', '/api/tags/tag1', None, Response.TAG_NOT_FOUND], + ['get', '/api/bookmarks/1', None, Response.BOOKMARK_NOT_FOUND], + ['put', '/api/bookmarks/1', {'title': 'none'}, Response.FAILURE], + ['delete', '/api/bookmarks/1', None, Response.FAILURE], + ['post', '/api/bookmarks/1/refresh', None, Response.FAILURE], + ['get', '/api/bookmarks/1/tiny', None, Response.FAILURE], + ['get', '/api/bookmarks/1/2', None, Response.RANGE_NOT_VALID], + ['put', '/api/bookmarks/1/2', {1: {'title': 'one'}, 2: {'title': 'two'}}, Response.RANGE_NOT_VALID], + ['delete', '/api/bookmarks/1/2', None, Response.RANGE_NOT_VALID], + ] +) +def test_invalid_id(client, method, url, json, exp_res): + rd = getattr(client, method)(url, json=json) + assert_response(rd, exp_res) def test_tag_api(client): url = 'http://google.com' - rd = client.post('/api/bookmarks', data={'url': url, 'tags': 'tag1,tag2'}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + rd = client.post('/api/bookmarks', json={'url': url, 'tags': ['tag1', 'TAG2']}) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/tags') - assert rd.status_code == 200 - assert rd.get_json() == {'tags': ['tag1', 'tag2']} + assert_response(rd, Response.SUCCESS, {'tags': ['tag1', 'tag2']}) rd = client.get('/api/tags/tag1') - assert rd.status_code == 200 - assert rd.get_json() == {'name': 'tag1', 'usage_count': 1} - rd = client.put('/api/tags/tag1', data={'tags': 'tag3,tag4'}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS, {'name': 'tag1', 'usage_count': 1}) + rd = client.put('/api/tags/tag1', json={'tags': 'string'}) + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': {'tags': 'List of tags expected.'}}) + for json in [{}, {'tags': None}, {'tags': ''}, {'tags':[]}]: + rd = client.put('/api/tags/tag1', json={'tags': []}) + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': {'tags': [['This field is required.']]}}) + rd = client.put('/api/tags/tag1', json={'tags': ['ok', '', None]}) + errors = {'tags': [[], ['This field is required.'], ['This field is required.']]} + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': errors}) + rd = client.put('/api/tags/tag1', json={'tags': ['one,two', 3,]}) + errors = {'tags': [['Tag must not contain delimiter \",\".'], ['Tag must be a string.']]} + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': errors}) + rd = client.put('/api/tags/tag1', json={'tags': ['tag3', 'TAG 4']}) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/tags') - assert rd.status_code == 200 - assert rd.get_json() == {'tags': ['tag2', 'tag3 tag4']} - rd = client.put('/api/tags/tag2', data={'tags': 'tag5'}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS, {'tags': ['tag 4', 'tag2', 'tag3']}) + rd = client.put('/api/tags/tag 4', json={'tags': ['tag5']}) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/tags') - assert rd.status_code == 200 - assert rd.get_json() == {'tags': ['tag3 tag4', 'tag5']} + assert_response(rd, Response.SUCCESS, {'tags': ['tag2', 'tag3', 'tag5']}) + rd = client.delete('/api/tags/tag3') + assert_response(rd, Response.SUCCESS) + rd = client.delete('/api/tags/tag3') + assert_response(rd, Response.TAG_NOT_FOUND) + rd = client.delete('/api/tags/tag,2') + assert_response(rd, Response.TAG_NOT_VALID) rd = client.get('/api/bookmarks/1') - assert rd.status_code == 200 - assert rd.get_json() == { - 'description': '', 'tags': ['tag3 tag4', 'tag5'], 'title': 'Google', - 'url': 'http://google.com'} + assert_response(rd, Response.SUCCESS, {'description': '', 'tags': ['tag2', 'tag5'], 'title': 'Google', 'url': url}) def test_bookmark_api(client): url = 'http://google.com' - rd = client.post('/api/bookmarks', data={'url': url}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] - rd = client.post('/api/bookmarks', data={'url': url}) - assert rd.status_code == 400 - assert rd.get_json() == response_template['failure'] + rd = client.post('/api/bookmarks', json={}) + errors = {'url': ['This field is required.']} + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': errors}) + rd = client.post('/api/bookmarks', json={'url': url}) + assert_response(rd, Response.SUCCESS) + rd = client.post('/api/bookmarks', json={'url': url}) + assert_response(rd, Response.FAILURE) rd = client.get('/api/bookmarks') - assert rd.status_code == 200 - assert rd.get_json() == {'bookmarks': [{ - 'description': '', 'tags': [], 'title': 'Google', 'url': 'http://google.com'}]} + assert_response(rd, Response.SUCCESS, {'bookmarks': [{'description': '', 'tags': [], 'title': 'Google', 'url': url}]}) rd = client.get('/api/bookmarks/1') - assert rd.status_code == 200 - assert rd.get_json() == { - 'description': '', 'tags': [], 'title': 'Google', 'url': 'http://google.com'} - rd = client.put('/api/bookmarks/1', data={'tags': [',tag1,tag2,']}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS, {'description': '', 'tags': [], 'title': 'Google', 'url': url}) + rd = client.put('/api/bookmarks/1', json={'tags': 'not a list'}) + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': {'tags': 'List of tags expected.'}}) + rd = client.put('/api/bookmarks/1', json={'tags': ['tag1', 'tag2']}) + assert_response(rd, Response.SUCCESS) + rd = client.put('/api/bookmarks/1', json={}) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks/1') - assert rd.status_code == 200 - assert rd.get_json() == { - 'description': '', 'tags': ['tag1', 'tag2'], 'title': 'Google', 'url': 'http://google.com'} + assert_response(rd, Response.SUCCESS, {'description': '', 'tags': ['tag1', 'tag2'], 'title': 'Google', 'url': url}) + rd = client.put('/api/bookmarks/1', json={'tags': [], 'description': 'Description'}) + assert_response(rd, Response.SUCCESS) + rd = client.get('/api/bookmarks/1') + assert_response(rd, Response.SUCCESS, {'description': 'Description', 'tags': [], 'title': 'Google', 'url': url}) @pytest.mark.parametrize('d_url', ['/api/bookmarks', '/api/bookmarks/1']) def test_bookmark_api_delete(client, d_url): url = 'http://google.com' - rd = client.post('/api/bookmarks', data={'url': url}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + rd = client.post('/api/bookmarks', json={'url': url}) + assert_response(rd, Response.SUCCESS) rd = client.delete(d_url) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS) @pytest.mark.parametrize('api_url', ['/api/bookmarks/refresh', '/api/bookmarks/1/refresh']) def test_refresh_bookmark(client, api_url): url = 'http://google.com' - rd = client.post('/api/bookmarks', data={'url': url}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + rd = client.post('/api/bookmarks', json={'url': url}) + assert_response(rd, Response.SUCCESS) rd = client.post(api_url) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks/1') - assert rd.status_code == 200 - json_data = rd.get_json() - json_data.pop('description') - assert json_data == {'tags': [], 'title': 'Google', 'url': 'http://google.com'} + assert_response(rd, Response.SUCCESS, {'description': '', 'tags': [], 'title': 'Google', 'url': url}) @pytest.mark.parametrize( - 'url, exp_res, status_code', [ - ['http://google.com', {'url': 'http://tny.im/2'}, 200], - ['chrome://bookmarks/', response_template['failure'], 400], + 'url, exp_res, data', [ + ['http://google.com', Response.SUCCESS, {'url': 'http://tny.im/2'}], + ['chrome://bookmarks/', Response.FAILURE, None], ]) -def test_get_tiny_url(client, url, exp_res, status_code): - rd = client.post('/api/bookmarks', data={'url': url}) - assert rd.status_code == 200 - assert rd.get_json() == response_template['success'] +def test_get_tiny_url(client, url, exp_res, data): + rd = client.post('/api/bookmarks', json={'url': url}) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks/1/tiny') - assert rd.status_code == status_code - assert rd.get_json() == exp_res + assert_response(rd, exp_res, data) -@pytest.mark.parametrize('kwargs, status_code, exp_res', [ +@pytest.mark.parametrize('kwargs, exp_res, data', [ [ {"data": {'url': 'http://google.com'}}, - 200, - { - 'bad url': 0, 'recognized mime': 0, - 'tags': None, 'title': 'Google'} + Response.SUCCESS, + {'bad url': 0, 'recognized mime': 0, 'tags': None, 'title': 'Google'} ], - [{}, 400, response_template['failure']], + [{}, Response.FAILURE, None], [ {"data": {'url': 'chrome://bookmarks/'}}, - 200, - { - 'bad url': 1, 'recognized mime': 0, - 'tags': None, 'title': None} + Response.SUCCESS, + {'bad url': 1, 'recognized mime': 0, 'tags': None, 'title': None} ], ]) -def test_network_handle(client, kwargs, status_code, exp_res): +def test_network_handle(client, kwargs, exp_res, data): rd = client.post('/api/network_handle', **kwargs) - assert rd.status_code == status_code + assert rd.status_code == exp_res.status_code rd_json = rd.get_json() rd_json.pop('description', None) - assert rd_json == exp_res + assert rd_json == exp_res.json(data=data) def test_bookmark_range_api(client): - status_code = 200 kwargs_list = [ - {"data": {'url': 'http://google.com'}}, - {"data": {'url': 'http://example.com'}}] + {"json": {'url': 'http://google.com'}}, + {"json": {'url': 'http://example.com'}}] for kwargs in kwargs_list: rd = client.post('/api/bookmarks', **kwargs) - assert rd.status_code == status_code + assert_response(rd, Response.SUCCESS) + + rd = client.put('/api/bookmarks/1/2', json={ + '1': {'tags': ['tag1 A', 'tag1 B', 'tag1 C']}, + '2': {'tags': ['tag2']} + }) + assert_response(rd, Response.SUCCESS) + rd = client.get('/api/bookmarks/1/2') + assert_response(rd, Response.SUCCESS, {'bookmarks': { + '1': {'description': '', 'tags': ['tag1 a', 'tag1 b', 'tag1 c'], 'title': 'Google', 'url': 'http://google.com'}, + '2': {'description': '', 'tags': ['tag2',], 'title': 'Example Domain', 'url': 'http://example.com'}}}) + rd = client.put('/api/bookmarks/1/2', json={ + '1': {'title': 'Bookmark 1', 'tags': ['tag1 C', 'tag1 A'], 'del_tags': True}, + '2': {'title': 'Bookmark 2', 'tags': ['-', 'tag2'], 'del_tags': False} + }) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks/1/2') - assert rd.status_code == status_code - assert rd.get_json() == { - 'bookmarks': { - '1': {'description': '', 'tags': [], 'title': 'Google', 'url': 'http://google.com'}, - '2': {'description': '', 'tags': [], 'title': 'Example Domain', 'url': 'http://example.com'}}} - put_data = json.dumps({1: {'tags': 'tag1'}, 2: {'tags': 'tag2'}}) - headers = {'content-type': 'application/json'} - rd = client.put('/api/bookmarks/1/2', data=put_data, headers=headers) - assert rd.status_code == status_code - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS, {'bookmarks': { + '1': {'description': '', 'tags': ['tag1 b'], 'title': 'Bookmark 1', 'url': 'http://google.com'}, + '2': {'description': '', 'tags': ['-', 'tag2',], 'title': 'Bookmark 2', 'url': 'http://example.com'}}}) + + rd = client.put('/api/bookmarks/2/1', json={}) + assert_response(rd, Response.RANGE_NOT_VALID) + + rd = client.put('/api/bookmarks/1/2', json={}) + assert_response(rd, Response.INPUT_NOT_VALID, data={ + 'errors': { + '1': 'Input required.', + '2': 'Input required.' + } + }) + rd = client.put('/api/bookmarks/1/2', json={'1': {'tags': []}}) + assert_response(rd, Response.INPUT_NOT_VALID, data={'errors': {'2': 'Input required.'}}) + rd = client.put('/api/bookmarks/1/2', json={ + '1': {'tags': ['ok', 'with,delim']}, + '2': {'tags': 'string'}, + }) + assert_response(rd, Response.INPUT_NOT_VALID, data={ + 'errors': { + '1': {'tags': [[], ['Tag must not contain delimiter \",\".']]}, + '2': {'tags': 'List of tags expected.'} + } + }) + rd = client.get('/api/bookmarks/2/1') + assert_response(rd, Response.RANGE_NOT_VALID) rd = client.delete('/api/bookmarks/1/2') - assert rd.status_code == status_code - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks') - assert rd.get_json() == {'bookmarks': []} + assert_response(rd, Response.SUCCESS, {'bookmarks': []}) def test_bookmark_search(client): - status_code = 200 - rd = client.post('/api/bookmarks', data={'url': 'http://google.com'}) - assert rd.status_code == status_code - assert rd.get_json() == response_template['success'] + rd = client.post('/api/bookmarks', json={'url': 'http://google.com'}) + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks/search', query_string={'keywords': ['google']}) - assert rd.status_code == status_code - assert rd.get_json() == {'bookmarks': [ - {'description': '', 'id': 1, 'tags': [], 'title': 'Google', 'url': 'http://google.com'}]} + assert_response(rd, Response.SUCCESS, {'bookmarks': [ + {'description': '', 'id': 1, 'tags': [], 'title': 'Google', 'url': 'http://google.com'}]}) rd = client.delete('/api/bookmarks/search', data={'keywords': ['google']}) - assert rd.status_code == status_code - assert rd.get_json() == response_template['success'] + assert_response(rd, Response.SUCCESS) rd = client.get('/api/bookmarks') - assert rd.get_json() == {'bookmarks': []} + assert_response(rd, Response.SUCCESS, {'bookmarks': []}) @pytest.mark.parametrize('env_val, exp_val', [