source: main/waeup.ikoba/trunk/src/waeup/ikoba/documents/tests/test_batching.py @ 12260

Last change on this file since 12260 was 12257, checked in by Henrik Bettermann, 10 years ago

Add batch processor tests.
File size: 8.0 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 12057 2014-11-25 13:15:27Z henrik $
3##
4## Copyright (C) 2014 Uli Fouquet & Henrik Bettermann
5## This program is free software; you can redistribute it and/or modify
6## it under the terms of the GNU General Public License as published by
7## the Free Software Foundation; either version 2 of the License, or
8## (at your option) any later version.
9##
10## This program is distributed in the hope that it will be useful,
11## but WITHOUT ANY WARRANTY; without even the implied warranty of
12## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13## GNU General Public License for more details.
14##
15## You should have received a copy of the GNU General Public License
16## along with this program; if not, write to the Free Software
17## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18##
19"""Unit tests for document data processors.
20"""
21import os
22import shutil
23import tempfile
24import unittest
25import datetime
26import grok
27from time import time
28from zope.event import notify
29from zope.component import createObject, queryUtility
30from zope.component.hooks import setSite, clearSite
31from zope.catalog.interfaces import ICatalog
32from zope.interface.verify import verifyClass, verifyObject
33from zope.securitypolicy.interfaces import IPrincipalRoleManager
34
35from waeup.ikoba.app import Company
36from waeup.ikoba.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
37from waeup.ikoba.documents.batching import PDFDocumentProcessor
38from waeup.ikoba.documents.document import PDFDocument, HTMLDocument
39from waeup.ikoba.testing import FunctionalLayer, FunctionalTestCase
40
41DOCUMENT_SAMPLE_DATA = open(
42    os.path.join(os.path.dirname(__file__), 'sample_document_data.csv'),
43    'rb').read()
44
45DOCUMENT_HEADER_FIELDS = DOCUMENT_SAMPLE_DATA.split(
46    '\n')[0].split(',')
47
48DOCUMENT_SAMPLE_DATA_UPDATE = open(
49    os.path.join(os.path.dirname(__file__), 'sample_document_data_update.csv'),
50    'rb').read()
51
52DOCUMENT_HEADER_FIELDS_UPDATE = DOCUMENT_SAMPLE_DATA_UPDATE.split(
53    '\n')[0].split(',')
54
55class DocumentImportExportSetup(FunctionalTestCase):
56
57    layer = FunctionalLayer
58
59    def setUp(self):
60        super(DocumentImportExportSetup, self).setUp()
61        self.dc_root = tempfile.mkdtemp()
62        self.workdir = tempfile.mkdtemp()
63        app = Company()
64        app['datacenter'].setStoragePath(self.dc_root)
65        self.getRootFolder()['app'] = app
66        self.app = self.getRootFolder()['app']
67        setSite(app)
68
69        self.logfile = os.path.join(
70            self.app['datacenter'].storage, 'logs', 'main.log')
71        return
72
73    def tearDown(self):
74        super(DocumentImportExportSetup, self).tearDown()
75        shutil.rmtree(self.workdir)
76        shutil.rmtree(self.dc_root)
77        clearSite()
78        return
79
80    def setup_for_export(self):
81        document = PDFDocument()
82        self.app['documents'][document.document_id] = self.document = document
83        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
84        role_manager = IPrincipalRoleManager(document)
85        role_manager.assignRoleToPrincipal(u'johnsrole', u'john')
86        return
87
88
89class PDFDocumentProcessorTest(DocumentImportExportSetup):
90
91    layer = FunctionalLayer
92
93    def setUp(self):
94        super(PDFDocumentProcessorTest, self).setUp()
95
96        # Add document
97        document = PDFDocument()
98        document.document_id = u'DOC1'
99        self.app['documents'].addDocument(document)
100        document.title = u'Our PDF Document'
101        notify(grok.ObjectModifiedEvent(document))
102        self.document = self.app['documents'][document.document_id]
103
104        self.processor = PDFDocumentProcessor()
105        self.csv_file = os.path.join(self.workdir, 'sample_document_data.csv')
106        self.csv_file_update = os.path.join(
107            self.workdir, 'sample_document_data_update.csv')
108        open(self.csv_file, 'wb').write(DOCUMENT_SAMPLE_DATA)
109        open(self.csv_file_update, 'wb').write(DOCUMENT_SAMPLE_DATA_UPDATE)
110
111    def test_interface(self):
112        # Make sure we fulfill the interface contracts.
113        assert verifyObject(IBatchProcessor, self.processor) is True
114        assert verifyClass(
115            IBatchProcessor, PDFDocumentProcessor) is True
116
117    def test_parentsExist(self):
118        self.assertFalse(self.processor.parentsExist(None, dict()))
119        self.assertTrue(self.processor.parentsExist(None, self.app))
120
121    def test_entryExists(self):
122        assert self.processor.entryExists(
123            dict(document_id='ID_NONE'), self.app) is False
124        assert self.processor.entryExists(
125            dict(document_id=self.document.document_id), self.app) is True
126
127    def test_getParent(self):
128        parent = self.processor.getParent(None, self.app)
129        assert parent is self.app['documents']
130
131    def test_getEntry(self):
132        assert self.processor.getEntry(
133            dict(document_id='ID_NONE'), self.app) is None
134        assert self.processor.getEntry(
135            dict(document_id=self.document.document_id), self.app) is self.document
136
137    def test_addEntry(self):
138        new_document = PDFDocument()
139        new_document.document_id = u'DOC2'
140        self.processor.addEntry(
141            new_document, {}, self.app)
142        assert len(self.app['documents'].keys()) == 2
143        self.assertEqual(self.app['documents']['DOC2'].document_id, 'DOC2')
144
145    def test_checkConversion(self):
146        errs, inv_errs, conv_dict = self.processor.checkConversion(
147            dict(document_id='DOC4', class_name='PDFDocument'))
148        self.assertEqual(len(errs),0)
149        errs, inv_errs, conv_dict = self.processor.checkConversion(
150            dict(document_id='id with spaces', class_name='PDFDocument'))
151        self.assertEqual(len(errs),1)
152        errs, inv_errs, conv_dict = self.processor.checkConversion(
153            dict(document_id='DOC4', class_name='WrongDocument'), mode='create')
154        self.assertEqual(len(errs),1)
155
156    def test_delEntry(self):
157        assert self.document.document_id in self.app['documents'].keys()
158        self.processor.delEntry(
159            dict(document_id=self.document.document_id), self.app)
160        assert self.document.document_id not in self.app['documents'].keys()
161
162    def test_import(self):
163        num, num_warns, fin_file, fail_file = self.processor.doImport(
164            self.csv_file, DOCUMENT_HEADER_FIELDS)
165        self.assertEqual(num_warns,2)
166        self.assertEqual(len(self.app['documents']), 4)
167        self.assertEqual(self.app['documents']['ABC'].title,'ABC title')
168        logcontent = open(self.logfile).read()
169        # Logging message from updateEntry
170        self.assertTrue(
171            'INFO - system - Public PDF Document Processor - sample_document_data - '
172            'EFG - updated: document_id=EFG, title=EFG title\n'
173            in logcontent)
174        failcontent = open(fail_file).read()
175        self.assertTrue(
176            'PDFDocument,ABC,Duplicate document,This object already exists.'
177            in failcontent)
178        self.assertTrue(
179            'HTMLDocument,HIJ,HIJ title,class_name: wrong processor'
180            in failcontent)
181        shutil.rmtree(os.path.dirname(fin_file))
182
183    def test_import_update(self):
184        num, num_warns, fin_file, fail_file = self.processor.doImport(
185            self.csv_file, DOCUMENT_HEADER_FIELDS)
186        shutil.rmtree(os.path.dirname(fin_file))
187        num, num_warns, fin_file, fail_file = self.processor.doImport(
188            self.csv_file_update, DOCUMENT_HEADER_FIELDS_UPDATE, 'update')
189        self.assertEqual(num_warns,1)
190        # title has changed
191        self.assertEqual(self.app['documents']['ABC'].title,'ABC new title')
192        logcontent = open(self.logfile).read()
193        # Logging message from updateEntry
194        self.assertTrue(
195            'INFO - system - Public PDF Document Processor - sample_document_data_update - '
196            'ABC - updated: document_id=ABC, title=ABC new title\n'
197            in logcontent)
198        failcontent = open(fail_file).read()
199        self.assertTrue(
200            'XYZ,Non-existing document,Cannot update: no such entry'
201            in failcontent)
202        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.