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

Last change on this file since 9271 was 9271, checked in by uli, 12 years ago

Shorten overlong lines.
  • Property svn:keywords set to Id
File size: 9.0 KB
Line 
1## $Id: test_batching.py 9271 2012-10-02 11:14:59Z uli $
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,
147            'initialized')
148        self.assertEqual(
149            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].batch_serial,
150            33)
151        self.assertEqual(
152            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].random_num,
153            '1625368961')
154        self.assertEqual(
155            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state, 'used')
156        self.assertMatches(
157            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
158            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
159            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system")
160        self.assertMatches(
161            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
162            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
163            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system")
164        logcontent = open(self.logfile).read()
165        self.assertTrue(
166            'INFO - system - AccessCode Processor - CLR-1-1625368961 - '
167            'AC updated: state=used, batch_serial=33, random_num=1625368961, '
168            'cost=100.0, owner=K1000009'
169            in logcontent)
170        shutil.rmtree(os.path.dirname(fin_file))
171        return
172
173    def test_import_update(self):
174        self.processor = AccessCodeBatchProcessor()
175        self.csv_file = os.path.join(self.workdir, 'sample_batch_data.csv')
176        open(self.csv_file, 'wb').write(BATCH_SAMPLE_DATA)
177        self.processor.doImport(self.csv_file, BATCH_HEADER_FIELDS)
178
179        self.processor = AccessCodeProcessor()
180        self.csv_file = os.path.join(self.workdir, 'sample_ac_data.csv')
181        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA)
182        num, num_warns, fin_file, fail_file =  self.processor.doImport(
183            self.csv_file, AC_HEADER_FIELDS)
184
185        self.assertEqual(self.app['accesscodes']['HOS-1'].used_num, 1)
186        self.assertEqual(self.app['accesscodes']['HOS-1'].disabled_num, 0)
187        self.assertEqual(self.app['accesscodes']['CLR-1'].used_num, 1)
188        self.assertEqual(self.app['accesscodes']['CLR-1'].disabled_num, 0)
189
190        self.csv_file = os.path.join(self.workdir, 'sample_ac_data_update.csv')
191        open(self.csv_file, 'wb').write(AC_SAMPLE_DATA_UPDATE)
192        num, num_warns, fin_file, fail_file =  self.processor.doImport(
193            self.csv_file, AC_HEADER_FIELDS_UPDATE, 'update')
194        fin_file = open(fin_file).read()
195        self.assertEqual(num_warns,0)
196        self.assertEqual(
197            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].state,
198            'disabled')
199        self.assertEqual(
200            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].state,
201            'disabled')
202        self.assertMatches(
203            self.app['accesscodes']['CLR-1']['CLR-1-1625368961'].history,
204            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
205            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'used' set by system|"
206            "|<YYYY-MM-DD hh:mm:ss> UTC - state 'disabled' set by system")
207        self.assertMatches(
208            self.app['accesscodes']['HOS-1']['HOS-1-98769876'].history,
209            "<YYYY-MM-DD hh:mm:ss> UTC - initialized by system|"
210            "|<YYYY-MM-DD hh:mm:ss> UTC - used by system|"
211            "|<YYYY-MM-DD hh:mm:ss> UTC - disabled by system")
212        # When importing transitions the counters are properly set.
213        self.assertEqual(self.app['accesscodes']['HOS-1'].used_num, 0)
214        self.assertEqual(self.app['accesscodes']['HOS-1'].disabled_num, 1)
215        # When importing states the counters remain unchanged.
216        self.assertEqual(self.app['accesscodes']['CLR-1'].used_num, 1)
217        self.assertEqual(self.app['accesscodes']['CLR-1'].disabled_num, 0)
218        return
Note: See TracBrowser for help on using the repository browser.