testShaDir.py

# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2011 Nexedi SA and Contributors. All Rights Reserved.
#                    Lucas Carvalho <lucas@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################


import httplib
import urlparse
import json
import transaction
import random
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from ShaDirMixin import ShaDirMixin
from Products.ERP5Type.tests.backportUnittest import expectedFailure

class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
  """
    ShaDir - HTTP Information Cache server
  """

  def getTitle(self):
    """
      Return the title of the current test set.
    """
    return "SHADIR - HTTP Information Cache Server"

  def postInformation(self, key=None, data=None):
    """
      Post the information calling the Python Script.
      It simulates the real usage.
    """
    parsed = urlparse.urlparse(self.shadir_url)
    connection = httplib.HTTPConnection(parsed.hostname, parsed.port)
    try:
      connection.request('PUT', '/'.join([parsed.path, key or self.key]),
        data or self.data, self.header_dict)
      result = connection.getresponse()
      data = result.read()
    finally:
      connection.close()
    self.assertEqual(result.status, httplib.CREATED)
    self.assertEqual(data, '')

  def getInformation(self, key=None):
    """
      Get the information calling the Python Script.
      It simulates the real usage.
    """
    parsed = urlparse.urlparse(self.shadir_url)
    connection = httplib.HTTPConnection(parsed.hostname, parsed.port)
    try:
      connection.request('GET', '/'.join([parsed.path, key or self.key]),
        self.data, self.header_dict)
      result = connection.getresponse()
      data = result.read()
    finally:
      connection.close()
    return result.status, data

  def beforeTearDown(self):
    """
      Clear everything for next test.
    """
    for module in ('data_set_module',
                   'document_module',):
      folder = self.portal[module]
      folder.manage_delObjects(list(folder.objectIds()))
    self.portal.portal_caches.clearAllCache()
    transaction.commit()
    self.tic()

  def test_post_information(self):
    """
      Check if posting information is working.
    """
    self.postInformation()
    transaction.commit()
    self.tic()

    # Asserting Data Set
    data_set = self.portal.portal_catalog.getResultValue(
      reference=self.key)
    self.assertEquals(self.key, data_set.getReference())
    self.assertEquals('published', data_set.getValidationState())

    # Asserting Document
    document = self.portal.portal_catalog.getResultValue(
      reference=self.sha512sum)
    self.assertEquals(self.sha512sum, document.getTitle())
    self.assertEquals(self.sha512sum, document.getReference())
    self.assertEquals(self.data, document.getData())
    self.assertEquals(data_set, document.getFollowUpValue())
    self.assertEquals(str(self.expiration_date),
                                    str(document.getExpirationDate()))
    self.assertEquals('application/json', document.getContentType())
    self.assertEquals('Published', document.getValidationStateTitle())

  def test_get_information(self):
    """
      check if return the temp document with text content.
    """
    self.postInformation()

    transaction.commit()
    self.tic()

    result, data = self.getInformation()
    self.assertEqual(result, httplib.OK)

    information_list = json.loads(data)

    self.assertEquals(1, len(information_list))
    self.assertEquals(json.dumps(information_list[0]), self.data)

  def test_post_information_more_than_once(self):
    """
      Check if posting information is working.
    """
    self.postInformation()
    transaction.commit()
    self.tic()

    self.postInformation()
    transaction.commit()
    self.tic()

    self.assertEqual(1, self.portal.portal_catalog.countResults(
      reference=self.key)[0][0])
    data_set = self.portal.portal_catalog.getResultValue(
      reference=self.key)
    self.assertEqual(self.key, data_set.getReference())
    self.assertEqual('published', data_set.getValidationState())

    document_list = data_set.getFollowUpRelatedValueList()

    self.assertEqual([self.sha512sum, self.sha512sum], [q.getReference() for q \
        in document_list])
    self.assertEqual(sorted(['published', 'archived']), sorted([
        q.getValidationState() for q in document_list]))

    result, data = self.getInformation()
    self.assertEqual(result, httplib.OK)
    information_list = json.loads(data)

    self.assertEquals(1, len(information_list))
    self.assertEquals(json.dumps(information_list[0]), self.data)

  @expectedFailure
  def test_post_information_more_than_once_no_tic(self):
    """
      Check if posting information is working.
    """
    self.postInformation()
    transaction.commit()

    self.postInformation()
    transaction.commit()
    self.tic()

    self.assertEqual(1, self.portal.portal_catalog.countResults(
      reference=self.key)[0][0])
    data_set = self.portal.portal_catalog.getResultValue(
      reference=self.key)
    self.assertEqual(self.key, data_set.getReference())
    self.assertEqual('published', data_set.getValidationState())

    document_list = data_set.getFollowUpRelatedValueList()

    self.assertEqual([self.sha512sum, self.sha512sum], [q.getReference() for q \
        in document_list])
    self.assertEqual(sorted(['published', 'archived']), sorted([
        q.getValidationState() for q in document_list]))

  def test_get_information_for_single_data_set(self):
    """
      check if return the temp document with text content.
    """
    self.postInformation()
    transaction.commit()
    self.tic()

    result, data = self.getInformation()
    self.assertEqual(result, httplib.OK)
    information_list = json.loads(data)

    self.assertEquals(1, len(information_list))
    self.assertEquals(json.dumps(information_list[0]), self.data)

  def test_get_information_from_different_data_set(self):
    """
      POST information with two different keys
      It must create two Data Set and two Text documents.

      When the user retrieve the content of a given key,
      it must return only the Text document related to the key.

      This relation is controlled by Data Set object.
    """
    self.postInformation()
    transaction.commit()
    self.tic()

    urlmd5_2 = 'anotherurlmd5' + str(random.random())
    sha512_2 = 'anothersha512_2' + str(random.random())
    key_2 = 'another_key' + str(random.random())
    data_list_2 = [json.dumps({'file': self.file_name,
                      'urlmd5': urlmd5_2,
                      'sha512': sha512_2,
                      'creation_date': str(self.creation_date),
                      'expiration_date': str(self.expiration_date),
                      'distribution': self.distribution,
                      'architecture': self.architecture}),
                      "User SIGNATURE goes here."]
    data_2 = json.dumps(data_list_2)
    self.postInformation(key_2, data_2)
    transaction.commit()
    self.tic()

    self.assertEquals(2, len(self.portal.data_set_module))
    self.assertEquals(2, len(self.portal.document_module))

    result, document = self.getInformation()
    self.assertEquals(1, len(json.loads(document)))

    result, document2 = self.getInformation(key_2)
    self.assertEquals(1, len(json.loads(document2)))

    self.postInformation()
    transaction.commit()
    self.tic()
    self.assertEquals(2, len(self.portal.data_set_module))
    self.assertEquals(3, len(self.portal.document_module))

    result, document3 = self.getInformation()
    self.assertEquals(1, len(json.loads(document3)))