tests/app/melange/logic/test_cached_list.py

# 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)