source: main/waeup.kofa/trunk/src/waeup/kofa/accesscodes/tests/test_batching.py @ 9266

Last change on this file since 9266 was 9266, checked in by Henrik Bettermann, 12 years ago

Repair and optimize AccessCodeProcessor?. AC states and transitions can now be imported in create and update mode.
  • Property svn:keywords set to Id
File size: 8.2 KB
Line 
1## $Id: test_batching.py 9266 2012-10-02 06:43:52Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18"""
19Tests for hostels and their UI components.
20"""
21import os
22import shutil
23import tempfile
24import grok
25import pytz
26from datetime import datetime, timedelta
27from zope.event import notify
28from zope.interface.verify import verifyClass, verifyObject
29from zope.component.hooks import setSite, clearSite
30from zope.testbrowser.testing import Browser
31from zope.security.interfaces import Unauthorized
32from zope.catalog.interfaces import ICatalog
33from zope.component import queryUtility
34from waeup.kofa.app import University
35from waeup.kofa.interfaces import IObjectHistory, IKofaPluggable, IKofaUtils
36from waeup.kofa.testing import (
37    FunctionalLayer, FunctionalTestCase, setUp, tearDown, getRootFolder)
38from waeup.kofa.accesscodes.batching import (
39    AccessCodeBatchProcessor, AccessCodeProcessor)
40
41
42BATCH_SAMPLE_DATA = open(
43    os.path.join(os.path.dirname(__file__), 'sample_batch_data.csv'),
44    'rb').read()
45
46BATCH_HEADER_FIELDS = BATCH_SAMPLE_DATA.split(
47    '\n')[0].split(',')
48
49AC_SAMPLE_DATA = open(
50    os.path.join(os.path.dirname(__file__), 'sample_ac_data.csv'),
51    'rb').read()
52
53AC_HEADER_FIELDS = AC_SAMPLE_DATA.split(
54    '\n')[0].split(',')
55
56AC_SAMPLE_DATA_UPDATE = open(
57    os.path.join(os.path.dirname(__file__), 'sample_ac_data_update.csv'),
58    'rb').read()
59
60AC_HEADER_FIELDS_UPDATE = AC_SAMPLE_DATA_UPDATE.split(
61    '\n')[0].split(',')
62
63class ACFullSetup(FunctionalTestCase):
64
65    def setUp(self):
66        super(ACFullSetup, self).setUp()
67
68        # Setup a sample site for each test
69        app = University()
70        self.dc_root = tempfile.mkdtemp()
71        app['datacenter'].setStoragePath(self.dc_root)
72
73        # Prepopulate the ZODB...
74        self.getRootFolder()['app'] = app
75        # we add the site immediately after creation to the
76        # ZODB. Catalogs and other local utilities are not setup
77        # before that step.
78        self.app = self.getRootFolder()['app']
79        # Set site here. Some of the following setup code might need
80        # to access grok.getSite() and should get our new app then
81        setSite(app)
82
83        # Put the prepopulated site into test ZODB and prepare test
84        # browser
85        self.browser = Browser()
86        self.browser.handleErrors = False
87
88        self.logfile = os.path.join(
89            self.app['datacenter'].storage, 'logs', 'accesscodes.log')
90        self.workdir = tempfile.mkdtemp()
91
92    def tearDown(self):
93        super(ACFullSetup, self).tearDown()
94        clearSite()
95        shutil.rmtree(self.dc_root)
96
97class ACBatchProcessorTest(ACFullSetup):
98
99    layer = FunctionalLayer
100
101    def test_import(self):
102        self.processor = AccessCodeBatchProcessor()
103        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
104        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
105        num, num_warns, fin_file, fail_file = self.processor.doImport(
106            self.csv_file, BATCH_HEADER_FIELDS)
107        self.assertEqual(num_warns,0)
108        self.assertEqual(len(self.app['accesscodes'].keys()), 7)
109        self.assertEqual(self.app['accesscodes']['CLR-1'].num,1)
110        logcontent = open(self.logfile).read()
111        self.assertTrue(
112            'system - AccessCodeBatch Processor - CLR-1 - '
113            'Batch updated: num=1, creator=system, used_num=1, '
114            'entry_num=3, creation_date=2012-04-28 07:28:48.719026+00:00, '
115            'prefix=CLR, cost=0.0, disabled_num=0'
116            in logcontent)
117        shutil.rmtree(os.path.dirname(fin_file))
118        return
119
120class ACProcessorTest(ACFullSetup):
121
122    layer = FunctionalLayer
123
124    def test_import_create(self):
125        self.processor = AccessCodeBatchProcessor()
126        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
127        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
128        self.processor.doImport(self.csv_file, BATCH_HEADER_FIELDS)
129
130        self.processor = AccessCodeProcessor()
131        self.csv_file = os.path.join(self.workdir, 'sample_ac_data.csv')
132        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA)
133        num, num_warns, fin_file, fail_file =  self.processor.doImport(
134            self.csv_file, AC_HEADER_FIELDS)
135        self.assertEqual(num_warns,3)
136        fail_file = open(fail_file).read()
137        self.assertTrue(
138            'HOS-1-0007044547,HOS,anything,state: not allowed' in fail_file)
139        self.assertTrue(
140            'HOS-1-555,HOS,anything,workflow: not allowed' in fail_file)
141        self.assertTrue(
142            'HOS-1-666,HOS,anything,transition: not allowed' in fail_file)
143        self.assertEqual(
144            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state, 'used')
145        self.assertEqual(
146            self.app['accesscodes']['HOS-1']['HOS-1-76254765'].state, 'initialized')
147        self.assertEqual(
148            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].batch_serial, 33)
149        self.assertEqual(
150            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].random_num, '1625368961')
151        self.assertEqual(
152            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state, 'used')
153        self.assertMatches(
154            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
155            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
156            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system")
157        self.assertMatches(
158            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
159            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
160            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system")
161        logcontent = open(self.logfile).read()
162        self.assertTrue(
163            'INFO - system - AccessCode Processor - CLR-1-1625368961 - '
164            'AC updated: state=used, batch_serial=33, random_num=1625368961, '
165            'cost=100.0, owner=K1000009'
166            in logcontent)
167        shutil.rmtree(os.path.dirname(fin_file))
168        return
169
170    def test_import_update(self):
171        self.processor = AccessCodeBatchProcessor()
172        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
173        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
174        self.processor.doImport(self.csv_file, BATCH_HEADER_FIELDS)
175
176        self.processor = AccessCodeProcessor()
177        self.csv_file = os.path.join(self.workdir, 'sample_ac_data.csv')
178        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA)
179        num, num_warns, fin_file, fail_file =  self.processor.doImport(
180            self.csv_file, AC_HEADER_FIELDS)
181
182        self.csv_file = os.path.join(self.workdir, 'sample_ac_data_update.csv')
183        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA_UPDATE)
184        num, num_warns, fin_file, fail_file =  self.processor.doImport(
185            self.csv_file, AC_HEADER_FIELDS_UPDATE, 'update')
186        fin_file = open(fin_file).read()
187        self.assertEqual(num_warns,0)
188        self.assertEqual(
189            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state, 'disabled')
190        self.assertEqual(
191            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state, 'disabled')
192        self.assertMatches(
193            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
194            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
195            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system|"
196            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'disabled' set by system")
197        self.assertMatches(
198            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
199            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
200            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system|"
201            "|<YYYY-MM-DD hh:mm:ss> UTC - disabled by system")
202        return
Note: See TracBrowser for help on using the repository browser.