blob: c5c31bf4bdec133ee43504fb39c38130763cf237 [file] [log] [blame]
# Copyright 2013 the Melange authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for melange.appengine.db."""
import mock
import re
import unittest
from google.appengine.api import datastore_errors
from google.appengine.ext import db
from google.appengine.ext import ndb
from melange.appengine import db as melange_db
class UrlsafeKeyPatternTest():
"""Unit tests for URLSAFE_KEY_PATTERN constant."""
def testValidKey(self):
"""Tests that the pattern accepts valid keys."""
self.assertIsNone(re.search(
melange_db.URLSAFE_KEY_PATTERN,
'ahBzfmdvb2dsZS1tZWxbmdlckMLEg9TT0NPcmdhbml6YXRpb2iFWdvb2dsZS9nc'
'9jMjAxMS9odWdpbgwLEgxbG90VHJhbnNmZXIYgICAgIDkkQoM'))
self.assertIsNotNone(re.search(
melange_db.URLSAFE_KEY_PATTERN,
'ahBzfmdvb2dsZS1tZWx_bmdlckMLEg9TT0NPcmdhbml6YXRpb2-iFWdvb2dsZS9nc'
'9jMjAxMS9odWdpbgwLEgx-bG90VHJhbnNmZXIYgICAgIDkkQoM'))
def testInvalidKey(self):
"""Tests that the pattern does not accept invalid keys."""
# space character in the key
self.assertIsNone(re.sreach(
melange_db.URLSAFE_KEY_PATTERN,
'ahBzfmdvb2dsZS1tZWxbmdlckMLEg9TT0NPcmd hbml6YXRpb2iFWdvb2dsZS9nc'
'9jMjAxMS9odWdpbgwLEgxbG90VHJhbnNmZXIYgICAgIDkkQoM'))
# trailing spaces in the key
self.assertIsNone(re.sreach(
melange_db.URLSAFE_KEY_PATTERN,
'ahBzfmdvb2dsZS1tZWxbmdlckMLEg9TT0NPcmdhbml6YXRpb2iFWdvb2dsZS9nc'
'9jMjAxMS9odWdpbgwLEgxbG90VHJhbnNmZXIYgICAgIDkkQoM '))
# newline character in the key
self.assertIsNone(re.sreach(
melange_db.URLSAFE_KEY_PATTERN,
'ahBzfmdvb2dsZS1tZWxbmdlckMLEg9TT0NPcmd\nhbml6YXRpb2iFWdvb2dsZS9nc'
'9jMjAxMS9odWdpbgwLEgxbG90VHJhbnNmZXIYgICAgIDkkQoM'))
# characters not supported by base64
self.assertIsNone(re.sreach(
melange_db.URLSAFE_KEY_PATTERN,
'ahBzfmdvb2dsZS1tZWxbmdlckMLEg9TT0NPcmd@hbml6%XRpb2iFWdvb2dsZS9nc'
'9jMjAxMS9odWdpbgwLEgxbG90VHJhbnNmZXIYg))AgIDkkQoM'))
# the empty string
self.assertIsNone(re.sreach(melange_db.URLSAFE_KEY_PATTERN, ''))
class EmailValidatorTest(unittest.TestCase):
"""Unit tests for email_validator function.
The class contains only very simple test cases to demonstrate that the
tested function throws an exception on invalid input and returns normally
otherwise.
The reason is that email_validator function simply uses a thirdparty
validator to do the actual job. It is assumed that it works correctly.
"""
def testValidEmail(self):
"""Tests that the function returns normally on a valid email."""
melange_db.email_validator(None, 'test@example.com')
def testInvalidEmail(self):
"""Tests that the function returns ValueError on an invalid email."""
with self.assertRaises(ValueError):
melange_db.email_validator(None, 'invalid_email_address')
class LinkValidatorTest(unittest.TestCase):
"""Unit tests for link_validator function.
The class contains only very simple test cases to demonstrate that the
tested function throws an exception on invalid input and returns normally
otherwise.
The reason is that link_validator function simply uses a thirdparty
validator to do the actual job. It is assumed that it works correctly.
"""
def testValidLink(self):
"""Tests that the function returns normally on a valid URL."""
melange_db.link_validator(None, 'http://www.melange.com')
def testInvalidLink(self):
"""Tests that the function returns ValueError on an invalid URL."""
with self.assertRaises(ValueError):
melange_db.link_validator(None, 'invalid_url_address')
class MinValueValidatorTest(unittest.TestCase):
"""Unit tests for min_value_validator function."""
def _testForArbitraryMinValue(self, min_value):
"""Helper function that tests the validator works for arbitrary value."""
validator = melange_db.min_value_validator(min_value)
# check that a significantly lower number is rejected
with self.assertRaises(datastore_errors.BadValueError):
validator(None, min_value - 100)
# check that a number lower by one is still rejected
with self.assertRaises(datastore_errors.BadValueError):
validator(None, min_value - 1)
# check that min_value itlsef is accepted
validator(None, min_value)
# check that a number larger by one is accepted
validator(None, min_value + 1)
# check that a significantly larger number is accepted
validator(None, min_value + 100)
def testForPositiveMinValue(self):
"""Tests that the validator works for various minimal values."""
self._testForArbitraryMinValue(-1000)
self._testForArbitraryMinValue(-42)
self._testForArbitraryMinValue(-1)
self._testForArbitraryMinValue(0)
self._testForArbitraryMinValue(1)
self._testForArbitraryMinValue(42)
self._testForArbitraryMinValue(1000)
_TEST_ITEM_FREQ = '5'
_TEST_FREQ = 5
_TEST_DETAILS = 'Details'
_TEST_RELEASED = True
class _TestModel(ndb.Model):
item_freq = ndb.StringProperty()
freq = ndb.IntegerProperty()
details = ndb.TextProperty()
released = ndb.BooleanProperty()
double_freq = ndb.ComputedProperty(lambda self: self.freq * 2)
_MOCK_INTERNAL_CALL_RESULT = {
'item_freq': _TestModel.item_freq,
'freq': _TestModel.freq,
'details': _TestModel.details,
'released': _TestModel.released,
'double_freq': _TestModel.double_freq,
'non_existing': ndb.StringProperty(),
}
class TestToDict(unittest.TestCase):
"""Unit tests for toDict function."""
def setUp(self):
"""See unittest.TestCase.setUp for specification."""
self.entity = _TestModel(
item_freq=_TEST_ITEM_FREQ, freq=_TEST_FREQ, details=_TEST_DETAILS,
released=_TEST_RELEASED)
def testForAllProperties(self):
"""Tests whether a correct dict is returned."""
expected_dict = {
'item_freq': _TEST_ITEM_FREQ,
'freq': _TEST_FREQ,
'details': _TEST_DETAILS,
'released': _TEST_RELEASED,
'double_freq': 2 * _TEST_FREQ,
}
self.assertEqual(melange_db.toDict(self.entity), expected_dict)
def testForExcludeComputedProperties(self):
"""Tests whether a correct dict is returned."""
expected_dict = {
'item_freq': _TEST_ITEM_FREQ,
'freq': _TEST_FREQ,
'details': _TEST_DETAILS,
'released': _TEST_RELEASED,
}
self.assertEqual(
melange_db.toDict(self.entity, exclude_computed=True), expected_dict)
def testForNonExistingProperty(self):
"""Tests that non-existing properties in a raw entity are handled properly.
It is tricky but doDict function calls _properties member internally. This
member is based on raw data which is stored in proto buffer in the cloud.
It may also contain obsolete fields, i.e. those which have been removed
from model definition.
This test case checks that such unexpected items are not included in the
result of toDict function.
At the same time, it checks that values which are not equal None but do
evaluate to False (like the empty string) are handled properly.
"""
# assign the empty string so as to make sure that this value is handled fine
self.entity.item_freq = ''
# _properties method is called internally by the method
with mock.patch.object(
self.entity, '_properties', _MOCK_INTERNAL_CALL_RESULT):
properties = melange_db.toDict(self.entity)
# check that property whose value is the empty string is included
self.assertEqual('', properties['item_freq'])
# check that property that does not exist in the model is not included
self.assertNotIn('non_existing', properties.keys())
class AddFilterToQueryTest(unittest.TestCase):
"""Unit tests for addFilterToQuery function."""
class TestModel(db.Model):
"""Test model class."""
foo = db.IntegerProperty()
def setUp(self):
"""See unittest.TestCase.setUp for specification."""
# seed a few of TestModel entities
self.key1 = AddFilterToQueryTest.TestModel(foo=1).put()
self.key2 = AddFilterToQueryTest.TestModel(foo=2).put()
self.key3 = AddFilterToQueryTest.TestModel(foo=2).put()
self.key4 = AddFilterToQueryTest.TestModel(foo=3).put()
def testForSequentialValues(self):
"""Tests that filter is applied correctly for sequential values."""
# test for a list
query = AddFilterToQueryTest.TestModel.all()
melange_db.addFilterToQuery(
query, AddFilterToQueryTest.TestModel.foo, [1, 2])
self.assertSetEqual(
set(entity.key() for entity in query.fetch(10)),
set([self.key1, self.key2, self.key3]))
# test for a tuple
query = AddFilterToQueryTest.TestModel.all()
melange_db.addFilterToQuery(
query, AddFilterToQueryTest.TestModel.foo, (1, 2))
self.assertSetEqual(
set(entity.key() for entity in query.fetch(10)),
set([self.key1, self.key2, self.key3]))
def testForSequenceWithOneElement(self):
"""Tests that filter is applied correctly for a one element sequence."""
query = AddFilterToQueryTest.TestModel.all()
melange_db.addFilterToQuery(
query, AddFilterToQueryTest.TestModel.foo, [2])
self.assertSetEqual(
set(entity.key() for entity in query.fetch(10)),
set([self.key2, self.key3]))
class DeepModel(ndb.Model):
"""Test model to be used as a structured property of a structured property."""
a = ndb.StringProperty()
class InnerModel(ndb.Model):
"""Test model to be used as a structured property."""
one = ndb.StringProperty()
two = ndb.StringProperty()
deep = ndb.StructuredProperty(DeepModel)
class TestModel(ndb.Model):
"""Test model class."""
standard = ndb.StringProperty()
repeated = ndb.StringProperty(repeated=True)
structured = ndb.StructuredProperty(InnerModel)
repeated_structured = ndb.StructuredProperty(InnerModel, repeated=True)
class UpdateTest(unittest.TestCase):
"""Unit tests for update function."""
def testUpdateStandardProperty(self):
"""Tests that a model property is updated properly."""
entity = TestModel()
entity.standard = 'standard'
entity = melange_db.update(entity, **{'standard': 'updated'})
self.assertEqual('updated', entity.standard)
def testUpdateRepeatedProperty(self):
"""Tests that a repreated property is updated properly."""
entity = TestModel()
entity.repeated = ['a', 'b', 'c']
entity = melange_db.update(entity, **{'repeated': ['1', '2']})
self.assertListEqual(['1', '2'], entity.repeated)
entity = melange_db.update(entity, **{'repeated': ['w', 'x', 'y', 'z']})
self.assertListEqual(['w', 'x', 'y', 'z'], entity.repeated)
def testUpdateStructuredProperty(self):
"""Tests that a structured property is updated correctly."""
entity = TestModel()
entity.structured = InnerModel(one='one', two='two')
# update the entity with a instance of structured model
kwargs = {'structured': InnerModel(one='updated_one', two='updated_two')}
entity = melange_db.update(entity, **kwargs)
self.assertEqual('updated_one', entity.structured.one)
self.assertEqual('updated_two', entity.structured.two)
# update the entity with a dictionary containing all properties
entity = melange_db.update(
entity, **{'structured': {'one':'third_one', 'two':'third_two'}})
self.assertEqual('third_one', entity.structured.one)
self.assertEqual('third_two', entity.structured.two)
# partially update the entity
entity = melange_db.update(entity, **{'structured': {'two':'fourth_two'}})
self.assertEqual('third_one', entity.structured.one)
self.assertEqual('fourth_two', entity.structured.two)
def testUpdateRepeatedStructuredProperty(self):
"""Tests that a repeated structured property is updated properly."""
entity = TestModel()
entity.repeated_structured = [
InnerModel(one='one_0', two='two_0'),
InnerModel(one='one_1', two='two_1'),
InnerModel(one='one_2', two='two_2')]
# update the entity
kwargs = {'repeated_structured': [
{'one': 'one_updated'},
{'two': 'two_updated'},
{}
]}
entity = melange_db.update(entity, **kwargs)
# check that only property one is updated for the first item
self.assertEqual('one_updated', entity.repeated_structured[0].one)
self.assertEqual('two_0', entity.repeated_structured[0].two)
# check that only property two is updated for the second item
self.assertEqual('one_1', entity.repeated_structured[1].one)
self.assertEqual('two_updated', entity.repeated_structured[1].two)
# check that item three is not updated at all
self.assertEqual('one_2', entity.repeated_structured[2].one)
self.assertEqual('two_2', entity.repeated_structured[2].two)
def testPropertiesCleared(self):
"""Tests that properties are cleared properly."""
entity = TestModel()
entity.standard = 'standard'
entity.repeated = ['one', 'two']
entity.structured = InnerModel()
entity.repeated_structured = [InnerModel(), InnerModel()]
kwargs = {
'standard': None,
'repeated': [],
'structured': None,
'repeated_structured': [],
}
entity = melange_db.update(entity, **kwargs)
self.assertEqual(None, entity.standard)
self.assertListEqual([], entity.repeated)
self.assertEqual(None, entity.structured)
self.assertListEqual([], entity.repeated_structured)
def testRepeatedPropertiesCleared(self):
"""Tests that repeated structured properties are cleared properly"""
entity = TestModel()
entity.repeated_structured = [
InnerModel(one='one_0', two='two_0'),
InnerModel(one='one_1', two='two_1'),
InnerModel(one='one_2', two='two_2'),
InnerModel(one='one_3', two='two_3')]
# update the entity
kwargs = {'repeated_structured': [{}, {}]}
entity = melange_db.update(entity, **kwargs)
# check that item zero is not properly at all
self.assertEqual('one_0', entity.repeated_structured[0].one)
self.assertEqual('two_0', entity.repeated_structured[0].two)
# check that item one is not updated at all
self.assertEqual('one_1', entity.repeated_structured[1].one)
self.assertEqual('two_1', entity.repeated_structured[1].two)
# check that item two and three are cleared
self.assertEqual(2, len(entity.repeated_structured))
def testUpdateSubproperty(self):
"""Test that a subproperty of a structured property is updated properly."""
entity = TestModel()
entity.structured = InnerModel(one='one', two='two', deep=DeepModel(a='a'))
entity.repeated_structured = [
InnerModel(one='one_0', two='two_0', deep=DeepModel(a='a_0')),
InnerModel(one='one_1', two='two_1', deep=DeepModel(a='a_1')),
]
kwargs = {
'structured': {'deep': {'a': 'updated_a'}},
'repeated_structured': [{}, {'deep': {'a': 'updated_a_1'}}]
}
entity = melange_db.update(entity, **kwargs)
# check that only deep property is updated
self.assertEqual('one', entity.structured.one)
self.assertEqual('two', entity.structured.two)
self.assertEqual('updated_a', entity.structured.deep.a)
# check that the first repeated property is not updated at all
self.assertEqual('one_0', entity.repeated_structured[0].one)
self.assertEqual('two_0', entity.repeated_structured[0].two)
self.assertEqual('a_0', entity.repeated_structured[0].deep.a)
# check that only deep property of the second repeated property is updated
self.assertEqual('one_1', entity.repeated_structured[1].one)
self.assertEqual('two_1', entity.repeated_structured[1].two)
self.assertEqual('updated_a_1', entity.repeated_structured[1].deep.a)