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

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

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