andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request #1128 from adamcik/feature/validate-core-input

Browse Source 
Validate core inputs
This commit is contained in:
Stein Magnus Jodal 2015-04-16 07:45:17 +02:00
commit e265f5d673
14 changed files with 396 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/changelog.rst
View File

@ -13,9 +13,11 @@ Core API
- Calling :meth:`mopidy.core.library.LibraryController.search`` with ``kwargs``
as the query is no longer supported (PR: :issue:`1090`)
- Update core controllers to handle backend exceptions in all calls that rely
- Updated core controllers to handle backend exceptions in all calls that rely
on multiple backends. (Issue: :issue:`667`)
- Update core methods to do strict input checking. (Fixes: :issue:`#700`)
Models
------

23
mopidy/core/library.py
View File

@ -5,8 +5,7 @@ import logging
import operator
import urlparse
from mopidy.utils import deprecation
from mopidy.utils import deprecation, validation
logger = logging.getLogger(__name__)
@ -70,6 +69,9 @@ class LibraryController(object):
"""
if uri is None:
return self._roots()
elif not uri.strip():
return []
validation.check_uri(uri)
return self._browse(uri)
def _roots(self):
@ -111,6 +113,9 @@ class LibraryController(object):
.. versionadded:: 1.0
"""
validation.check_choice(field, validation.DISTINCT_FIELDS)
query is None or validation.check_query(query) # TODO: normalize?
result = set()
futures = {b: b.library.get_distinct(field, query)
for b in self.backends.with_library.values()}
@ -137,6 +142,8 @@ class LibraryController(object):
.. versionadded:: 1.0
"""
validation.check_uris(uris)
futures = {
backend: backend.library.get_images(backend_uris)
for (backend, backend_uris)
@ -187,6 +194,11 @@ class LibraryController(object):
if none_set or both_set:
raise ValueError("One of 'uri' or 'uris' must be set")
# TODO: validation.one_of(*args)?
uris is None or validation.check_uris(uris)
uri is None or validation.check_uri(uri)
if uri:
deprecation.warn('core.library.lookup:uri_arg')
@ -219,6 +231,8 @@ class LibraryController(object):
:param uri: directory or track URI
:type uri: string
"""
uri is None or validation.check_uri(uri)
futures = {}
backends = {}
uri_scheme = urlparse.urlparse(uri).scheme if uri else None
@ -285,6 +299,10 @@ class LibraryController(object):
"""
query = _normalize_query(query or kwargs)
uris is None or validation.check_uris(uris)
query is None or validation.check_query(query)
validation.check_boolean(exact)
if kwargs:
deprecation.warn('core.library.search:kwargs_query')
@ -319,6 +337,7 @@ class LibraryController(object):
def _normalize_query(query):
broken_client = False
# TODO: this breaks if query is not a dictionary like object...
for (field, values) in query.items():
if isinstance(values, basestring):
broken_client = True

5
mopidy/core/mixer.py
View File

@ -2,6 +2,8 @@ from __future__ import absolute_import, unicode_literals
import logging
from mopidy.utils import validation
logger = logging.getLogger(__name__)
@ -31,6 +33,8 @@ class MixerController(object):
Returns :class:`True` if call is successful, otherwise :class:`False`.
"""
validation.check_integer(volume, min=0, max=100)
if self._mixer is None:
return False
else:
@ -52,6 +56,7 @@ class MixerController(object):
Returns :class:`True` if call is successful, otherwise :class:`False`.
"""
validation.check_boolean(mute)
if self._mixer is None:
return False
else:

8
mopidy/core/playback.py
View File

@ -3,9 +3,10 @@ from __future__ import absolute_import, unicode_literals
import logging
import urlparse
from mopidy import models
from mopidy.audio import PlaybackState
from mopidy.core import listener
from mopidy.utils import deprecation
from mopidy.utils import deprecation, validation
logger = logging.getLogger(__name__)
@ -96,6 +97,8 @@ class PlaybackController(object):
"PAUSED" -> "PLAYING" [ label="resume" ]
"PAUSED" -> "STOPPED" [ label="stop" ]
"""
validation.check_choice(new_state, validation.PLAYBACK_STATES)
(old_state, self._state) = (self.get_state(), new_state)
logger.debug('Changing state: %s -> %s', old_state, new_state)
@ -270,6 +273,7 @@ class PlaybackController(object):
:param tl_track: track to play
:type tl_track: :class:`mopidy.models.TlTrack` or :class:`None`
"""
tl_track is None or validation.check_instance(tl_track, models.TlTrack)
self._play(tl_track, on_error_step=1)
def _play(self, tl_track=None, on_error_step=1):
@ -360,6 +364,8 @@ class PlaybackController(object):
:type time_position: int
:rtype: :class:`True` if successful, else :class:`False`
"""
validation.check_integer(time_position, min=0)
if not self.core.tracklist.tracks:
return False

17
mopidy/core/playlists.py
View File

@ -5,7 +5,7 @@ import urlparse
from mopidy.core import listener
from mopidy.models import Playlist
from mopidy.utils import deprecation
from mopidy.utils import deprecation, validation
logger = logging.getLogger(__name__)
@ -62,6 +62,8 @@ class PlaylistsController(object):
.. versionadded:: 1.0
"""
validation.check_uri(uri)
uri_scheme = urlparse.urlparse(uri).scheme
backend = self.backends.with_playlists.get(uri_scheme, None)
if backend:
@ -139,6 +141,8 @@ class PlaylistsController(object):
:param uri: URI of the playlist to delete
:type uri: string
"""
validation.check_uri(uri)
uri_scheme = urlparse.urlparse(uri).scheme
backend = self.backends.with_playlists.get(uri_scheme, None)
if backend:
@ -172,6 +176,10 @@ class PlaylistsController(object):
deprecation.warn('core.playlists.filter')
criteria = criteria or kwargs
validation.check_query(
criteria, validation.PLAYLIST_FIELDS, list_values=False)
# TODO: stop using self playlists
matches = self.playlists
for (key, value) in criteria.iteritems():
matches = filter(lambda p: getattr(p, key) == value, matches)
@ -207,6 +215,8 @@ class PlaylistsController(object):
:param uri_scheme: limit to the backend matching the URI scheme
:type uri_scheme: string
"""
# TODO: check: uri_scheme is None or uri_scheme?
futures = {}
backends = {}
playlists_loaded = False
@ -251,8 +261,11 @@ class PlaylistsController(object):
:type playlist: :class:`mopidy.models.Playlist`
:rtype: :class:`mopidy.models.Playlist` or :class:`None`
"""
validation.check_instance(playlist, Playlist)
if playlist.uri is None:
return
return # TODO: log this problem?
uri_scheme = urlparse.urlparse(playlist.uri).scheme
backend = self.backends.with_playlists.get(uri_scheme, None)
if backend:

49
mopidy/core/tracklist.py
View File

@ -1,13 +1,11 @@
from __future__ import absolute_import, unicode_literals
import collections
import logging
import random
from mopidy import compat
from mopidy.core import listener
from mopidy.models import TlTrack
from mopidy.utils import deprecation
from mopidy.models import TlTrack, Track
from mopidy.utils import deprecation, validation
logger = logging.getLogger(__name__)
@ -93,6 +91,7 @@ class TracklistController(object):
:class:`False`
Tracks are not removed from the tracklist.
"""
validation.check_boolean(value)
if self.get_consume() != value:
self._trigger_options_changed()
return setattr(self, '_consume', value)
@ -121,7 +120,7 @@ class TracklistController(object):
:class:`False`
Tracks are played in the order of the tracklist.
"""
validation.check_boolean(value)
if self.get_random() != value:
self._trigger_options_changed()
if value:
@ -157,7 +156,7 @@ class TracklistController(object):
:class:`False`
The tracklist is played once.
"""
validation.check_boolean(value)
if self.get_repeat() != value:
self._trigger_options_changed()
return setattr(self, '_repeat', value)
@ -188,6 +187,7 @@ class TracklistController(object):
:class:`False`
Playback continues after current song.
"""
validation.check_boolean(value)
if self.get_single() != value:
self._trigger_options_changed()
return setattr(self, '_single', value)
@ -205,9 +205,10 @@ class TracklistController(object):
The position of the given track in the tracklist.
:param tl_track: the track to find the index of
:type tl_track: :class:`mopidy.models.TlTrack`
:type tl_track: :class:`mopidy.models.TlTrack` or :class:`None`
:rtype: :class:`int` or :class:`None`
"""
tl_track is None or validation.check_instance(tl_track, TlTrack)
try:
return self._tl_tracks.index(tl_track)
except ValueError:
@ -223,6 +224,7 @@ class TracklistController(object):
:type tl_track: :class:`mopidy.models.TlTrack` or :class:`None`
:rtype: :class:`mopidy.models.TlTrack` or :class:`None`
"""
tl_track is None or validation.check_instance(tl_track, TlTrack)
if self.get_single() and self.get_repeat():
return tl_track
elif self.get_single():
@ -247,6 +249,7 @@ class TracklistController(object):
:type tl_track: :class:`mopidy.models.TlTrack` or :class:`None`
:rtype: :class:`mopidy.models.TlTrack` or :class:`None`
"""
tl_track is None or validation.check_instance(tl_track, TlTrack)
if not self.get_tl_tracks():
return None
@ -288,6 +291,8 @@ class TracklistController(object):
:type tl_track: :class:`mopidy.models.TlTrack` or :class:`None`
:rtype: :class:`mopidy.models.TlTrack` or :class:`None`
"""
tl_track is None or validation.check_instance(tl_track, TlTrack)
if self.get_repeat() or self.get_consume() or self.get_random():
return tl_track
@ -330,8 +335,13 @@ class TracklistController(object):
"""
assert tracks is not None or uri is not None or uris is not None, \
'tracks, uri or uris must be provided'
# TODO: check that only one of tracks uri and uris is set...
# TODO: can at_position be negative?
# TODO: assert that tracks are track instances
tracks is None or validation.check_instances(tracks, Track)
uri is None or validation.check_uri(uri)
uris is None or validation.check_uris(uris)
validation.check_integer(at_position or 0)
if tracks:
deprecation.warn('core.tracklist.add:tracks_arg')
@ -412,17 +422,18 @@ class TracklistController(object):
:rtype: list of :class:`mopidy.models.TlTrack`
"""
criteria = criteria or kwargs
tlids = criteria.pop('tlid', [])
validation.check_query(criteria, validation.TRACKLIST_FIELDS)
validation.check_instances(tlids, int)
# TODO: deprecate kwargs
# TODO: id=[1, 2, 3] filtering can't possibly be working
matches = self._tl_tracks
for (key, values) in criteria.items():
if (not isinstance(values, collections.Iterable) or
isinstance(values, compat.string_types)):
# Fail hard if anyone is using the <0.17 calling style
raise ValueError('Filter values must be iterable: %r' % values)
if key == 'tlid':
matches = [ct for ct in matches if ct.tlid in values]
else:
matches = [
ct for ct in matches if getattr(ct.track, key) in values]
matches = [
ct for ct in matches if getattr(ct.track, key) in values]
if tlids:
matches = [ct for ct in matches if ct.tlid in tlids]
return matches
def move(self, start, end, to_position):
@ -443,6 +454,7 @@ class TracklistController(object):
tl_tracks = self._tl_tracks
# TODO: use validation helpers?
assert start < end, 'start must be smaller than end'
assert start >= 0, 'start must be at least zero'
assert end <= len(tl_tracks), \
@ -470,6 +482,7 @@ class TracklistController(object):
:type criteria: dict
:rtype: list of :class:`mopidy.models.TlTrack` that was removed
"""
# TODO: deprecate kwargs
tl_tracks = self.filter(criteria, **kwargs)
for tl_track in tl_tracks:
position = self._tl_tracks.index(tl_track)
@ -491,6 +504,7 @@ class TracklistController(object):
"""
tl_tracks = self._tl_tracks
# TOOD: use validation helpers?
if start is not None and end is not None:
assert start < end, 'start must be smaller than end'
@ -519,6 +533,7 @@ class TracklistController(object):
:type end: int
:rtype: :class:`mopidy.models.TlTrack`
"""
# TODO: validate slice?
return self._tl_tracks[start:end]
def _mark_playing(self, tl_track):

4
mopidy/exceptions.py
View File

@ -46,3 +46,7 @@ class ScannerError(MopidyException):
class AudioException(MopidyException):
pass
class ValidationError(ValueError):
pass

9
mopidy/mpd/protocol/current_playlist.py
View File

@ -1,5 +1,7 @@
from __future__ import absolute_import, unicode_literals
import urlparse
from mopidy.mpd import exceptions, protocol, translator
from mopidy.utils import deprecation
@ -21,8 +23,11 @@ def add(context, uri):
if not uri.strip('/'):
return
if context.core.tracklist.add(uris=[uri]).get():
return
# If we have an URI just try and add it directly without bothering with
# jumping through browse...
if urlparse.urlparse(uri).scheme != '':
if context.core.tracklist.add(uris=[uri]).get():
return
try:
uris = []

4
mopidy/mpd/protocol/music_db.py
View File

@ -270,10 +270,12 @@ def list_(context, *args):
if field not in _LIST_MAPPING:
raise exceptions.MpdArgError('incorrect arguments')
query = None
if len(params) == 1:
if field != 'album':
raise exceptions.MpdArgError('should be "Album" for 3 arguments')
query = {'artist': params}
if params[0].strip():
query = {'artist': params}
else:
try:
query = _query_from_mpd_search_parameters(params, _LIST_MAPPING)

102
mopidy/utils/validation.py Normal file
View File

@ -0,0 +1,102 @@
from __future__ import absolute_import, unicode_literals
import collections
import urlparse
from mopidy import compat, exceptions
PLAYBACK_STATES = {'paused', 'stopped', 'playing'}
SEARCH_FIELDS = {
'uri', 'track_name', 'album', 'artist', 'albumartist', 'composer',
'performer', 'track_no', 'genre', 'date', 'comment', 'any'}
PLAYLIST_FIELDS = {'uri', 'name'} # TODO: add length and last_modified?
TRACKLIST_FIELDS = { # TODO: add bitrate, length, disc_no, track_no, modified?
'uri', 'name', 'genre', 'date', 'comment', 'musicbrainz_id'}
DISTINCT_FIELDS = {
'artist', 'albumartist', 'album', 'composer', 'performer', 'date', 'genre'}
# TODO: _check_iterable(check, msg, **kwargs) + [check(a) for a in arg]?
def _check_iterable(arg, msg, **kwargs):
"""Ensure we have an iterable which is not a string."""
if isinstance(arg, compat.string_types):
raise exceptions.ValidationError(msg.format(arg=arg, **kwargs))
elif not isinstance(arg, collections.Iterable):
raise exceptions.ValidationError(msg.format(arg=arg, **kwargs))
def check_choice(arg, choices, msg='Expected one of {choices}, not {arg!r}'):
if arg not in choices:
raise exceptions.ValidationError(msg.format(
arg=arg, choices=tuple(choices)))
def check_boolean(arg, msg='Expected a boolean, not {arg!r}'):
check_instance(arg, bool, msg=msg)
def check_instance(arg, cls, msg='Expected a {name} instance, not {arg!r}'):
if not isinstance(arg, cls):
raise exceptions.ValidationError(
msg.format(arg=arg, name=cls.__name__))
def check_instances(arg, cls, msg='Expected a list of {name}, not {arg!r}'):
_check_iterable(arg, msg, name=cls.__name__)
if not all(isinstance(instance, cls) for instance in arg):
raise exceptions.ValidationError(
msg.format(arg=arg, name=cls.__name__))
def check_integer(arg, min=None, max=None):
if not isinstance(arg, (int, long)):
raise exceptions.ValidationError('Expected an integer, not %r' % arg)
elif min is not None and arg < min:
raise exceptions.ValidationError(
'Expected number larger or equal to %d, not %r' % (min, arg))
elif max is not None and arg > max:
raise exceptions.ValidationError(
'Expected number smaller or equal to %d, not %r' % (max, arg))
def check_query(arg, fields=SEARCH_FIELDS, list_values=True):
# TODO: normalize name -> track_name
# TODO: normalize value -> [value]
# TODO: normalize blank -> [] or just remove field?
# TODO: remove list_values?
if not isinstance(arg, collections.Mapping):
raise exceptions.ValidationError(
'Expected a query dictionary, not {arg!r}'.format(arg=arg))
for key, value in arg.items():
check_choice(key, fields, msg='Expected query field to be one of '
'{choices}, not {arg!r}')
if list_values:
msg = 'Expected "{key}" to be list of strings, not {arg!r}'
_check_iterable(value, msg, key=key)
[_check_query_value(key, v, msg) for v in value]
else:
_check_query_value(
key, value, 'Expected "{key}" to be a string, not {arg!r}')
def _check_query_value(key, arg, msg):
if not isinstance(arg, compat.string_types) or not arg.strip():
raise exceptions.ValidationError(msg.format(arg=arg, key=key))
def check_uri(arg, msg='Expected a valid URI, not {arg!r}'):
if not isinstance(arg, compat.string_types):
raise exceptions.ValidationError(msg.format(arg=arg))
elif urlparse.urlparse(arg).scheme == '':
raise exceptions.ValidationError(msg.format(arg=arg))
def check_uris(arg, msg='Expected a list of URIs, not {arg!r}'):
_check_iterable(arg, msg)
[check_uri(a, msg) for a in arg]

56
tests/local/test_library.py
View File

@ -9,7 +9,7 @@ import mock
import pykka
from mopidy import core
from mopidy import core, exceptions
from mopidy.local import actor, json
from mopidy.models import Album, Artist, Image, Track
@ -137,8 +137,8 @@ class LocalLibraryProviderTest(unittest.TestCase):
self.assertEqual(result[uri], self.tracks[0:1])
def test_lookup_unknown_track(self):
tracks = self.library.lookup(uris=['fake uri'])
self.assertEqual(tracks, {'fake uri': []})
tracks = self.library.lookup(uris=['fake:/uri'])
self.assertEqual(tracks, {'fake:/uri': []})
# test backward compatibility with local libraries returning a
# single Track
@ -343,42 +343,44 @@ class LocalLibraryProviderTest(unittest.TestCase):
result = self.find_exact(any=['local:track:path1'])
self.assertEqual(list(result[0].tracks), self.tracks[:1])
# TODO: This is really just a test of the query validation code now,
# as this code path never even makes it to the local backend.
def test_find_exact_wrong_type(self):
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(wrong=['test'])
def test_find_exact_with_empty_query(self):
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(artist=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(albumartist=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(track_name=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(composer=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(performer=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(album=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(track_no=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(genre=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(date=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(comment=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.find_exact(any=[''])
def test_search_no_hits(self):
@ -553,41 +555,41 @@ class LocalLibraryProviderTest(unittest.TestCase):
self.assertEqual(list(result[0].tracks), self.tracks[:1])
def test_search_wrong_type(self):
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(wrong=['test'])
def test_search_with_empty_query(self):
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(artist=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(albumartist=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(composer=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(performer=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(track_name=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(album=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(genre=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(date=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(comment=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(uri=[''])
with self.assertRaises(LookupError):
with self.assertRaises(exceptions.ValidationError):
self.search(any=[''])
def test_default_get_images_impl_no_images(self):

20
tests/local/test_playback.py
View File

@ -10,7 +10,7 @@ import pykka
from mopidy import core
from mopidy.core import PlaybackState
from mopidy.local import actor
from mopidy.models import Track
from mopidy.models import TlTrack, Track
from mopidy.utils import deprecation
from tests import dummy_audio, path_to_data_dir
@ -841,22 +841,6 @@ class LocalPlaybackProviderTest(unittest.TestCase):
self.playback.seek(self.tracklist.tracks[-1].length * 100)
self.assertEqual(self.playback.state, PlaybackState.STOPPED)
@unittest.SkipTest
@populate_tracklist
def test_seek_beyond_start_of_song(self):
# FIXME need to decide return value
self.playback.play()
result = self.playback.seek(-1000)
self.assert_(not result, 'Seek return value was %s' % result)
@populate_tracklist
def test_seek_beyond_start_of_song_update_postion(self):
self.playback.play()
self.playback.seek(-1000)
position = self.playback.time_position
self.assertGreaterEqual(position, 0)
self.assertEqual(self.playback.state, PlaybackState.PLAYING)
@populate_tracklist
def test_stop_when_stopped(self):
self.playback.stop()
@ -1088,4 +1072,4 @@ class LocalPlaybackProviderTest(unittest.TestCase):
@populate_tracklist
def test_playing_track_that_isnt_in_playlist(self):
with self.assertRaises(AssertionError):
self.playback.play((17, Track()))
self.playback.play(TlTrack(17, Track()))

6
tests/mpd/test_status.py
View File

@ -76,7 +76,7 @@ class StatusHandlerTest(unittest.TestCase):
self.assertEqual(int(result['repeat']), 0)
def test_status_method_contains_repeat_is_1(self):
self.core.tracklist.repeat = 1
self.core.tracklist.set_repeat(True)
result = dict(status.status(self.context))
self.assertIn('repeat', result)
self.assertEqual(int(result['repeat']), 1)
@ -87,7 +87,7 @@ class StatusHandlerTest(unittest.TestCase):
self.assertEqual(int(result['random']), 0)
def test_status_method_contains_random_is_1(self):
self.core.tracklist.random = 1
self.core.tracklist.set_random(True)
result = dict(status.status(self.context))
self.assertIn('random', result)
self.assertEqual(int(result['random']), 1)
@ -103,7 +103,7 @@ class StatusHandlerTest(unittest.TestCase):
self.assertEqual(int(result['consume']), 0)
def test_status_method_contains_consume_is_1(self):
self.core.tracklist.consume = 1
self.core.tracklist.set_consume(True)
result = dict(status.status(self.context))
self.assertIn('consume', result)
self.assertEqual(int(result['consume']), 1)

163
tests/utils/test_validation.py Normal file
View File

@ -0,0 +1,163 @@
from __future__ import absolute_import, unicode_literals
from pytest import raises
from mopidy import compat, exceptions
from mopidy.utils import validation
def test_check_boolean_with_valid_values():
for value in (True, False):
validation.check_boolean(value)
def test_check_boolean_with_other_values():
for value in 1, 0, None, '', list(), tuple():
with raises(exceptions.ValidationError):
validation.check_boolean(value)
def test_check_boolean_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_boolean(1234)
assert 'Expected a boolean, not 1234' == str(excinfo.value)
def test_check_choice_with_valid_values():
for value, choices in (2, (1, 2, 3)), ('abc', ('abc', 'def')):
validation.check_choice(value, choices)
def test_check_choice_with_invalid_values():
for value, choices in (5, (1, 2, 3)), ('xyz', ('abc', 'def')):
with raises(exceptions.ValidationError):
validation.check_choice(value, choices)
def test_check_choice_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_choice(5, (1, 2, 3))
assert 'Expected one of (1, 2, 3), not 5' == str(excinfo.value)
def test_check_instance_with_valid_choices():
for value, cls in ((True, bool), ('a', compat.text_type), (123, int)):
validation.check_instance(value, cls)
def test_check_instance_with_invalid_values():
for value, cls in (1, str), ('abc', int):
with raises(exceptions.ValidationError):
validation.check_instance(value, cls)
def test_check_instance_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_instance(1, dict)
assert 'Expected a dict instance, not 1' == str(excinfo.value)
def test_check_instances_with_valid_values():
validation.check_instances([], int)
validation.check_instances([1, 2], int)
validation.check_instances((1, 2), int)
def test_check_instances_with_invalid_values():
with raises(exceptions.ValidationError):
validation.check_instances('abc', compat.string_types)
with raises(exceptions.ValidationError):
validation.check_instances(['abc', 123], compat.string_types)
with raises(exceptions.ValidationError):
validation.check_instances(None, compat.string_types)
with raises(exceptions.ValidationError):
validation.check_instances([None], compat.string_types)
def test_check_instances_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_instances([1], compat.string_types)
assert 'Expected a list of basestring, not [1]' == str(excinfo.value)
def test_check_query_valid_values():
for value in {}, {'any': []}, {'any': ['abc']}:
validation.check_query(value)
def test_check_query_random_iterables():
for value in None, tuple(), list(), 'abc':
with raises(exceptions.ValidationError):
validation.check_query(value)
def test_check_mapping_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_query([])
assert 'Expected a query dictionary, not []' == str(excinfo.value)
def test_check_query_invalid_fields():
for value in 'wrong', 'bar', 'foo', 'tlid':
with raises(exceptions.ValidationError):
validation.check_query({value: []})
def test_check_field_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_query({'wrong': ['abc']})
assert 'Expected query field to be one of ' in str(excinfo.value)
def test_check_query_invalid_values():
for value in '', None, 'foo', 123, [''], [None]:
with raises(exceptions.ValidationError):
validation.check_query({'any': value})
def test_check_values_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_query({'any': 'abc'})
assert 'Expected "any" to be list of strings, not' in str(excinfo.value)
def test_check_uri_with_valid_values():
for value in 'foobar:', 'http://example.com', 'git+http://example.com':
validation.check_uri(value)
def test_check_uri_with_invalid_values():
# Note that tuple catches a potential bug with using "'foo' % arg" for
# formatting.
for value in ('foobar', 'htt p://example.com', None, 1234, tuple()):
with raises(exceptions.ValidationError):
validation.check_uri(value)
def test_check_uri_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_uri('testing')
assert "Expected a valid URI, not u'testing'" == str(excinfo.value)
def test_check_uris_with_valid_values():
validation.check_uris([])
validation.check_uris(['foobar:'])
validation.check_uris(('foobar:',))
def test_check_uris_with_invalid_values():
with raises(exceptions.ValidationError):
validation.check_uris('foobar:')
with raises(exceptions.ValidationError):
validation.check_uris(None)
with raises(exceptions.ValidationError):
validation.check_uris([None])
with raises(exceptions.ValidationError):
validation.check_uris(['foobar:', 'foobar'])
def test_check_uris_error_message():
with raises(exceptions.ValidationError) as excinfo:
validation.check_uris('testing')
assert "Expected a list of URIs, not u'testing'" == str(excinfo.value)