blob: 3c21043b357d6ab94d68f7dc6cf126ab38f8bfeb [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 functions in soc.logic.cached_list module."""
import datetime
import unittest
from melange import key_column_id_const
from melange.logic import cached_list as cached_list_logic
from melange.models import cached_list as cached_list_model
from soc.modules.seeder.logic.seeder import logic as seeder_logic
KEY = key_column_id_const.KEY_COLUMN_ID
class TestCacheItems(unittest.TestCase):
"""Test cacheItems function."""
def testAddingNewCachedList(self):
cached_list_logic.setCacheItems(
'foo_list', [{KEY: 'one', 'foo': 'bar'}, {KEY: 'two', 'baz': 'qux'}])
cached_list = cached_list_model.CachedList.query().get()
self.assertIsNotNone(cached_list)
def testAddingNewItems(self):
new_item1 = {KEY: 'one', 'name': 'foo'}
new_item2 = {KEY: 'two', 'name': 'bar'}
new_item3 = {KEY: 'three', 'name': 'baz'}
cached_list_logic.setCacheItems(
'foo_list', [new_item1, new_item2, new_item3])
cached_list = cached_list_model.CachedList.query().get()
self.assertIn(new_item1, cached_list.list_data)
self.assertIn(new_item2, cached_list.list_data)
def testUpdatingAfterCaching(self):
"""Tests whether cached list state is updated."""
valid_period = datetime.timedelta(2, 4, 6)
cached_list_logic.setCacheItems(
'test_list', [{KEY: 'foo'}, {KEY: 'bar'}], valid_period)
cached_list = cached_list_model.CachedList.get_by_id('test_list')
self.assertAlmostEqual(cached_list.valid_through,
datetime.datetime.now() + valid_period,
delta=datetime.timedelta(seconds=5))
self.assertFalse(cached_list.is_processing)
def testRemovingDuplicates(self):
"""Tests whether the duplicate list items are removed upon saving."""
item1 = {KEY: 'one', 'name': 'foo'}
item2 = {KEY: 'two', 'name': 'bar'}
item3 = {KEY: 'three', 'name': 'baz'}
dup_item1 = {KEY: 'one', 'name': 'foo'}
dup_item2 = {KEY: 'two', 'name': 'qux'}
list_with_duplicates = [item1, item2, item3, dup_item1, dup_item2]
# duplicate items should not be present in the cached list
expected_list = [item1, item2, item3]
cached_list_logic.setCacheItems('test_list', list_with_duplicates)
cached_list = cached_list_model.CachedList.get_by_id('test_list')
self.assertListEqual(cached_list.list_data, expected_list)
class TestGetCachedItems(unittest.TestCase):
"""Tests getCachedItems function."""
def setUp(self):
self.item1 = {'name': 'foo'}
self.item2 = {'name': 'bar'}
self.item3 = {'name': 'baz'}
self.item4 = {'name': 'qux'}
self.item5 = {'name': 'quux'}
cached_list_properties = {
'id': 'test_list',
'list_data':
[self.item1, self.item2, self.item3, self.item4, self.item5]
}
seeder_logic.seed(cached_list_model.CachedList, cached_list_properties)
def testRetrievingCachedItems(self):
cached_items = cached_list_logic.getCachedItems('test_list', 0, 5)
self.assertSequenceEqual(
[self.item1, self.item2, self.item3, self.item4, self.item5],
cached_items)
def testRetrievingPartOfCachedItems(self):
cached_items = cached_list_logic.getCachedItems('test_list', 1, 3)
self.assertSequenceEqual([self.item2, self.item3, self.item4], cached_items)
def testRetrievingWithOverSpecifiedLimit(self):
cached_items = cached_list_logic.getCachedItems('test_list', 2, 100)
self.assertSequenceEqual([self.item3, self.item4, self.item5], cached_items)
def testRetrievingWithOverSpecifiedStart(self):
cached_items = cached_list_logic.getCachedItems('test_list', 100, 3)
self.assertEqual(0, len(cached_items))
def testRetrievingWithoutSpecifyingStart(self):
"""Test whether 0 is taken as the starting index."""
cached_items = cached_list_logic.getCachedItems('test_list', limit=3)
self.assertListEqual([self.item1, self.item2, self.item3], cached_items)
def testRetrievingWithoutSpecifyingLimit(self):
"""Test whether all the items from the starting index is returned."""
cached_items = cached_list_logic.getCachedItems('test_list', start=2)
self.assertListEqual([self.item3, self.item4, self.item5], cached_items)
def testErrorForNonExsistentList(self):
with self.assertRaises(ValueError):
cached_list_logic.getCachedItems('none_existent', 0, 1)
class TestIsCachedListExists(unittest.TestCase):
"""Unit tests for isCachedListExists function."""
def setUp(self):
seeder_logic.seed(cached_list_model.CachedList, {'id': 'cached_list'})
def testForExistence(self):
"""Test whether True is returned only when the list exists."""
self.assertTrue(cached_list_logic.isCachedListExists('cached_list'))
self.assertFalse(cached_list_logic.isCachedListExists('none_existent'))
class TestCachedListStates(unittest.TestCase):
"""Unit tests for functions which check and set state of a CachedList."""
def setUp(self):
self.valid_list_id = 'valid_cached_list'
self.invalid_list_id = 'invalid_cached_list'
self.processing_list_id = 'processing_cached_list'
self.not_processing_list_id = 'not_processing_cached_list'
valid_cached_list_properties = {
'id': self.valid_list_id,
'valid_through': datetime.datetime.max
}
self.valid_list = seeder_logic.seed(
cached_list_model.CachedList, valid_cached_list_properties)
invalid_cached_list_properties = {
'id': self.invalid_list_id,
'valid_through': datetime.datetime.min
}
self.invalid_list = seeder_logic.seed(
cached_list_model.CachedList, invalid_cached_list_properties)
processing_cached_list_properties = {
'id': self.processing_list_id,
'is_processing': True
}
self.processing_list = seeder_logic.seed(
cached_list_model.CachedList, processing_cached_list_properties)
not_processing_cached_list_properties = {
'id': self.not_processing_list_id,
'is_processing': False
}
self.processing_list = seeder_logic.seed(
cached_list_model.CachedList, not_processing_cached_list_properties)
def testIsValid(self):
"""Tests isValid function."""
self.assertTrue(cached_list_logic.isValid(self.valid_list_id))
self.assertFalse(cached_list_logic.isValid(self.invalid_list_id))
def testIsProcessing(self):
"""Tests isProcessing function."""
self.assertTrue(cached_list_logic.isProcessing(self.processing_list_id))
self.assertFalse(
cached_list_logic.isProcessing(self.not_processing_list_id))
def testSetProcessing(self):
"""Tests setProcessing function."""
cached_list_logic.setProcessing(self.not_processing_list_id)
updated_list = cached_list_model.CachedList.get_by_id(
self.not_processing_list_id)
self.assertTrue(updated_list.is_processing)
class TestCreateEmptyProcessingList(unittest.TestCase):
"""Unit tests for createEmptyProcessingList function."""
def testCreateEmptyProcessingList(self):
"""Tests createEmptyProcessingList"""
cached_list_logic.createEmptyProcessingList('empty_processing_list')
test_list = cached_list_model.CachedList.get_by_id('empty_processing_list')
self.assertListEqual([], test_list.list_data)
self.assertTrue(test_list.is_processing)