source: main/waeup.kofa/branches/uli-py3/src/waeup/kofa/documents/tests/test_batching.py @ 17403

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

Propset svn:keywords "Id"
  • Property svn:keywords set to Id
File size: 8.3 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 12438 2015-01-11 08:27:37Z 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.kofa.app import University
36from waeup.kofa.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
37from waeup.kofa.documents.batching import PDFDocumentProcessor
38from waeup.kofa.documents.document import PDFDocument, HTMLDocument
39from waeup.kofa.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 = University()
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        document1 = PDFDocument()
82        document1.document_id = u'DOC1'
83        self.app['documents'].addDocument(document1)
84        self.pdfdocument = document1
85        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
86        role_manager = IPrincipalRoleManager(document1)
87        role_manager.assignRoleToPrincipal(u'johnsrole', u'john')
88
89        document2 = HTMLDocument()
90        document2.document_id = u'DOC2'
91        self.app['documents'].addDocument(document2)
92        self.htmldocument = document2
93        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
94        role_manager = IPrincipalRoleManager(document2)
95        role_manager.assignRoleToPrincipal(u'johnsrole', u'john')
96        return
97
98
99class PDFDocumentProcessorTest(DocumentImportExportSetup):
100
101    layer = FunctionalLayer
102
103    def setUp(self):
104        super(PDFDocumentProcessorTest, self).setUp()
105
106        # Add document
107        document = PDFDocument()
108        document.document_id = u'DOC1'
109        self.app['documents'].addDocument(document)
110        document.title = u'Our PDF Document'
111        notify(grok.ObjectModifiedEvent(document))
112        self.document = self.app['documents'][document.document_id]
113
114        self.processor = PDFDocumentProcessor()
115        self.csv_file = os.path.join(self.workdir, 'sample_document_data.csv')
116        self.csv_file_update = os.path.join(
117            self.workdir, 'sample_document_data_update.csv')
118        open(self.csv_file, 'wb').write(DOCUMENT_SAMPLE_DATA)
119        open(self.csv_file_update, 'wb').write(DOCUMENT_SAMPLE_DATA_UPDATE)
120
121    def test_interface(self):
122        # Make sure we fulfill the interface contracts.
123        assert verifyObject(IBatchProcessor, self.processor) is True
124        assert verifyClass(
125            IBatchProcessor, PDFDocumentProcessor) is True
126
127    def test_parentsExist(self):
128        self.assertFalse(self.processor.parentsExist(None, dict()))
129        self.assertTrue(self.processor.parentsExist(None, self.app))
130
131    def test_entryExists(self):
132        assert self.processor.entryExists(
133            dict(document_id='ID_NONE'), self.app) is False
134        assert self.processor.entryExists(
135            dict(document_id=self.document.document_id), self.app) is True
136
137    def test_getParent(self):
138        parent = self.processor.getParent(None, self.app)
139        assert parent is self.app['documents']
140
141    def test_getEntry(self):
142        assert self.processor.getEntry(
143            dict(document_id='ID_NONE'), self.app) is None
144        assert self.processor.getEntry(
145            dict(document_id=self.document.document_id), self.app) is self.document
146
147    def test_addEntry(self):
148        new_document = PDFDocument()
149        new_document.document_id = u'DOC2'
150        self.processor.addEntry(
151            new_document, {}, self.app)
152        assert len(self.app['documents'].keys()) == 2
153        self.assertEqual(self.app['documents']['DOC2'].document_id, 'DOC2')
154
155    def test_checkConversion(self):
156        errs, inv_errs, conv_dict = self.processor.checkConversion(
157            dict(document_id='DOC4', class_name='PDFDocument'))
158        self.assertEqual(len(errs),0)
159        errs, inv_errs, conv_dict = self.processor.checkConversion(
160            dict(document_id='id with spaces', class_name='PDFDocument'))
161        self.assertEqual(len(errs),1)
162        errs, inv_errs, conv_dict = self.processor.checkConversion(
163            dict(document_id='DOC4', class_name='WrongDocument'), mode='create')
164        self.assertEqual(len(errs),1)
165
166    def test_delEntry(self):
167        assert self.document.document_id in self.app['documents'].keys()
168        self.processor.delEntry(
169            dict(document_id=self.document.document_id), self.app)
170        assert self.document.document_id not in self.app['documents'].keys()
171
172    def test_import(self):
173        num, num_warns, fin_file, fail_file = self.processor.doImport(
174            self.csv_file, DOCUMENT_HEADER_FIELDS)
175        self.assertEqual(num_warns,2)
176        self.assertEqual(len(self.app['documents']), 4)
177        self.assertEqual(self.app['documents']['ABC'].title,'ABC title')
178        logcontent = open(self.logfile).read()
179        # Logging message from updateEntry
180        self.assertTrue(
181            'INFO - system - Public PDF Document Processor - sample_document_data - '
182            'EFG - updated: document_id=EFG, title=EFG title\n'
183            in logcontent)
184        failcontent = open(fail_file).read()
185        self.assertTrue(
186            'PDFDocument,ABC,Duplicate document,This object already exists.'
187            in failcontent)
188        self.assertTrue(
189            'HTMLDocument,HIJ,HIJ title,class_name: wrong processor'
190            in failcontent)
191        shutil.rmtree(os.path.dirname(fin_file))
192
193    def test_import_update(self):
194        num, num_warns, fin_file, fail_file = self.processor.doImport(
195            self.csv_file, DOCUMENT_HEADER_FIELDS)
196        shutil.rmtree(os.path.dirname(fin_file))
197        num, num_warns, fin_file, fail_file = self.processor.doImport(
198            self.csv_file_update, DOCUMENT_HEADER_FIELDS_UPDATE, 'update')
199        self.assertEqual(num_warns,1)
200        # title has changed
201        self.assertEqual(self.app['documents']['ABC'].title,'ABC new title')
202        logcontent = open(self.logfile).read()
203        # Logging message from updateEntry
204        self.assertTrue(
205            'INFO - system - Public PDF Document Processor - sample_document_data_update - '
206            'ABC - updated: document_id=ABC, title=ABC new title\n'
207            in logcontent)
208        failcontent = open(fail_file).read()
209        self.assertTrue(
210            'XYZ,Non-existing document,Cannot update: no such entry'
211            in failcontent)
212        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.