source: main/waeup.kofa/branches/0.2/src/waeup/kofa/accesscodes/tests/test_batching.py @ 14395

Last change on this file since 14395 was 10448, checked in by Henrik Bettermann, 11 years ago

Add TSC access code batch.
  • Property svn:keywords set to Id
File size: 9.3 KB
Line 
1## $Id: test_batching.py 10448 2013-08-05 06:07:53Z 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        shutil.rmtree(self.workdir)
97
98class ACBatchProcessorTest(ACFullSetup):
99
100    layer = FunctionalLayer
101
102    def test_import(self):
103        self.processor = AccessCodeBatchProcessor()
104        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
105        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
106        num, num_warns, fin_file, fail_file = self.processor.doImport(
107            self.csv_file, BATCH_HEADER_FIELDS)
108        self.assertEqual(num_warns,0)
109        self.assertEqual(len(self.app['accesscodes'].keys()), 8)
110        self.assertEqual(self.app['accesscodes']['CLR-1'].num,1)
111        logcontent = open(self.logfile).read()
112        self.assertTrue(
113            'system - AccessCodeBatch Processor - sample_batch_data - '
114            'CLR-1 - updated: '
115            'num=1, creator=system, used_num=1, '
116            'entry_num=3, creation_date=2012-04-28 07:28:48.719026+00:00, '
117            'prefix=CLR, cost=0.0, disabled_num=0'
118            in logcontent)
119        shutil.rmtree(os.path.dirname(fin_file))
120        return
121
122class ACProcessorTest(ACFullSetup):
123
124    layer = FunctionalLayer
125
126    def test_import_create(self):
127        self.processor = AccessCodeBatchProcessor()
128        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
129        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
130        num, num_warns, fin_file, fail_file =  self.processor.doImport(
131            self.csv_file, BATCH_HEADER_FIELDS)
132        shutil.rmtree(os.path.dirname(fin_file))
133
134        self.processor = AccessCodeProcessor()
135        self.csv_file = os.path.join(self.workdir, 'sample_ac_data.csv')
136        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA)
137        num, num_warns, fin_file, fail_file =  self.processor.doImport(
138            self.csv_file, AC_HEADER_FIELDS)
139        self.assertEqual(num_warns,3)
140        fail_file = open(fail_file).read()
141        self.assertTrue(
142            'HOS-1-0007044547,HOS,anything,state: not allowed' in fail_file)
143        self.assertTrue(
144            'HOS-1-555,HOS,anything,workflow: not allowed' in fail_file)
145        self.assertTrue(
146            'HOS-1-666,HOS,anything,transition: not allowed' in fail_file)
147        self.assertEqual(
148            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state, 'used')
149        self.assertEqual(
150            self.app['accesscodes']['HOS-1']['HOS-1-76254765'].state,
151            'initialized')
152        self.assertEqual(
153            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].batch_serial,
154            33)
155        self.assertEqual(
156            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].random_num,
157            '1625368961')
158        self.assertEqual(
159            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state, 'used')
160        self.assertMatches(
161            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
162            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
163            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system")
164        self.assertMatches(
165            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
166            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
167            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system")
168        logcontent = open(self.logfile).read()
169        self.assertTrue(
170            'INFO - system - AccessCode Processor - sample_ac_data - '
171            'CLR-1-1625368961 - updated: state=used, '
172            'batch_serial=33, random_num=1625368961, '
173            'cost=100.0, owner=K1000009'
174            in logcontent)
175        shutil.rmtree(os.path.dirname(fin_file))
176        return
177
178    def test_import_update(self):
179        self.processor = AccessCodeBatchProcessor()
180        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
181        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
182        num, num_warns, fin_file, fail_file =  self.processor.doImport(
183            self.csv_file, BATCH_HEADER_FIELDS)
184        shutil.rmtree(os.path.dirname(fin_file))
185
186        self.processor = AccessCodeProcessor()
187        self.csv_file = os.path.join(self.workdir, 'sample_ac_data.csv')
188        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA)
189        num, num_warns, fin_file, fail_file =  self.processor.doImport(
190            self.csv_file, AC_HEADER_FIELDS)
191        shutil.rmtree(os.path.dirname(fin_file))
192
193        self.assertEqual(self.app['accesscodes']['HOS-1'].used_num, 1)
194        self.assertEqual(self.app['accesscodes']['HOS-1'].disabled_num, 0)
195        self.assertEqual(self.app['accesscodes']['CLR-1'].used_num, 1)
196        self.assertEqual(self.app['accesscodes']['CLR-1'].disabled_num, 0)
197
198        self.csv_file = os.path.join(self.workdir, 'sample_ac_data_update.csv')
199        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA_UPDATE)
200        num, num_warns, fin_file, fail_file =  self.processor.doImport(
201            self.csv_file, AC_HEADER_FIELDS_UPDATE, 'update')
202        self.assertEqual(num_warns,0)
203        self.assertEqual(
204            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state,
205            'disabled')
206        self.assertEqual(
207            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state,
208            'disabled')
209        self.assertMatches(
210            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
211            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
212            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system|"
213            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'disabled' set by system")
214        self.assertMatches(
215            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
216            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
217            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system|"
218            "|<YYYY-MM-DD hh:mm:ss> UTC - disabled by system")
219        # When importing transitions the counters are properly set.
220        self.assertEqual(self.app['accesscodes']['HOS-1'].used_num, 0)
221        self.assertEqual(self.app['accesscodes']['HOS-1'].disabled_num, 1)
222        # When importing states the counters remain unchanged.
223        self.assertEqual(self.app['accesscodes']['CLR-1'].used_num, 1)
224        self.assertEqual(self.app['accesscodes']['CLR-1'].disabled_num, 0)
225        shutil.rmtree(os.path.dirname(fin_file))
226        return
Note: See TracBrowser for help on using the repository browser.