source: main/waeup.sirp/trunk/src/waeup/sirp/applicants/tests/test_batching.py @ 6658

##
## test_batching.py
## Login : <uli@pu.smp.net>
## Started on  Tue Aug 24 02:04:44 2010 Uli Fouquet
## $Id$
## 
## Copyright (C) 2010 Uli Fouquet
## 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
##
"""Unit tests for applicants-related data importers.
"""
import grok
import os
import shutil
import tempfile
import unittest
from zope.app.testing.functional import FunctionalTestCase
from zope.component.hooks import setSite, clearSite
from zope.interface.verify import verifyClass, verifyObject

from waeup.sirp.app import University
from waeup.sirp.applicants.batching import ApplicantsContainerImporter
from waeup.sirp.applicants.container import ApplicantsContainer
from waeup.sirp.applicants.interfaces import IApplicantsContainer
from waeup.sirp.interfaces import DuplicationError
from waeup.sirp.testing import FunctionalLayer, doctestsuite_for_module
from waeup.sirp.interfaces import IBatchProcessor


# Sample data we can use in tests...
APPS_CONTAINER_SAMPLE_DATA = open(
    os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'),
    'rb').read()

# The header fields of the above CSV snippet
APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split(
    '\n')[0].split(',')

class ApplicantsContainerImporterTest(FunctionalTestCase):

    layer = FunctionalLayer

    def setUp(self):
        super(ApplicantsContainerImporterTest, self).setUp()

        # Setup a sample site for each test
        app = University()
        self.dc_root = tempfile.mkdtemp()
        app['datacenter'].setStoragePath(self.dc_root)

        # Prepopulate the ZODB...
        self.getRootFolder()['app'] = app
        self.app = self.getRootFolder()['app']
        self.container = ApplicantsContainer()
        self.container.code = u'dp2011'
        self.app['applicants']['dp2011'] = self.container

        self.importer = ApplicantsContainerImporter()
        self.workdir = tempfile.mkdtemp()
        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
        setSite(self.app)
        return

    def tearDown(self):
        super(ApplicantsContainerImporterTest, self).tearDown()
        shutil.rmtree(self.workdir)
        shutil.rmtree(self.dc_root)
        clearSite()
        return

    def test_interface(self):
        # Make sure we fulfill the interface contracts.
        assert verifyObject(IBatchProcessor, self.importer) is True
        assert verifyClass(
            IBatchProcessor, ApplicantsContainerImporter) is True

    def test_parentsExist(self):
        assert self.importer.parentsExist(None, dict()) is False
        assert self.importer.parentsExist(None, self.app) is True

    def test_entryExists(self):
        assert self.importer.entryExists(
            dict(code='REG_NONE'), self.app) is False
        assert self.importer.entryExists(
            dict(code='dp2011'), self.app) is True

    def test_getParent(self):
        parent = self.importer.getParent(None, self.app)
        assert parent is self.app['applicants']

    def test_getEntry(self):
        assert self.importer.getEntry(
            dict(code='REG_NONE'), self.app) is None
        assert self.importer.getEntry(
            dict(code='dp2011'), self.app) is self.container

    def test_addEntry(self):
        self.importer.addEntry(
            'New application', dict(code='dp2012'), self.app)
        assert self.app['applicants']['dp2012'] == 'New application'

    def test_delEntry(self):
        self.importer.delEntry(dict(code='dp2011'), self.app)
        assert 'dp2011' not in self.app['applicants'].keys()

    def test_import(self):
        # Do a real import
        # see local sample_container.csv file for input
        num, num_warns, fin_file, fail_file = self.importer.doImport(
            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
        avail_containers = [x for x in self.app['applicants'].keys()]
        self.assertTrue(u'CODE1' in avail_containers)
        self.assertTrue(u'CODE2' in avail_containers)
        shutil.rmtree(os.path.dirname(fin_file))

def test_suite():
    suite = unittest.TestSuite()
    for testcase in [
        ApplicantsContainerImporterTest,
        ]:
        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
                testcase
                )
        )
    return suite