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..35e48f60 100644
--- a/bukuserver/response.py
+++ b/bukuserver/response.py
@@ -1,4 +1,47 @@
-response_template = {
- "success": {'status': 0, 'message': 'success'},
- "failure": {'status': 1, 'message': 'failure'}
-}
+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 = None) -> dict:
+ return dict(status=self.status, message=self.message, **data or {}) # pylint: disable=R1735
+
+ def __call__(self, *, data :dict = 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/<tag>', 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/<tag>', 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/<int:rec_id>', 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..06d8f358 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -1,14 +1,28 @@
-import json
-
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 = 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 +52,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', [