Changeset 11790 for main/waeup.kofa/trunk/src

Timestamp:

1 Sep 2014, 12:05:55 (10 years ago)

Author:

Henrik Bettermann

Message:

Extend and fix CertCourseProcessorTests?. Improve logging.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/university

Files:

• batching.py (modified) (1 diff)
• tests/sample_certificatecourse_data.csv (added)
• tests/test_batching.py (modified) (6 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/university/batching.py

@@ -353 +353 @@
     iface = ICertificateCourse
 
-    location_fields = ['course', 'level', 'faculty_code', 'department_code',
-                       'certificate_code',]
+    location_fields = ['certificate_code', 'course', 'level', 'faculty_code',
+                       'department_code',]
     factory_name = 'waeup.CertificateCourse'
 

main/waeup.kofa/trunk/src/waeup/kofa/university/tests/test_batching.py

@@ -81 +81 @@
     '\n')[0].split(',')
 
+
+CERTIFICATECOURSE_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_certificatecourse_data.csv'),
+    'rb').read()
+
+CERTIFICATECOURSE_HEADER_FIELDS = CERTIFICATECOURSE_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
 class UniversityProcessorSetup(FunctionalTestCase):
 
@@ -90 +98 @@
         self.workdir = tempfile.mkdtemp()
         app = University()
+        app['datacenter'].setStoragePath(self.dc_root)
         self.getRootFolder()['app'] = app
         self.app = self.getRootFolder()['app']
@@ -312 +321 @@
         # Logging message from updateEntry
         self.assertTrue(
-            'INFO - system - Department Processor - sample_department_data '
-            '- DEP2 - updated: local_roles=anne|waeup.local.DepartmentManager, '
-            'code=DEP2\n'
+            'INFO - system - Faculty Processor - sample_faculty_data_update - '
+            'FAC1 - updated: code=FAC1'
+            in logcontent)
+        self.assertTrue(
+            'INFO - system - Faculty Processor - sample_faculty_data_update - '
+            'FAC2 - updated: local_roles=bob|waeup.local.DepartmentManager, '
+            'code=FAC2'
+            in logcontent)
+        self.assertTrue(
+            'INFO - system - Faculty Processor - sample_faculty_data_update - '
+            'FAC3 - updated: code=FAC3'
+            in logcontent)
+        self.assertTrue(
+            'INFO - system - Faculty Processor - sample_faculty_data_update - '
+            'FAC4 - updated: code=FAC4'
             in logcontent)
         shutil.rmtree(os.path.dirname(fin_file))
@@ -636 +657 @@
 
 
-class CertCourseProcessorTests(FunctionalTestCase):
+class CertCourseProcessorTests(UniversityProcessorSetup):
 
     layer = FunctionalLayer
@@ -643 +664 @@
         super(CertCourseProcessorTests, self).setUp()
         self.proc = CertificateCourseProcessor()
-        self.site0 = dict()
-        self.site1 = dict(faculties=dict())
-        self.site2 = dict(faculties=dict(FAC=dict()))
-        self.department1 = Department(code='DPT')
-        self.department2 = Department(code='DPT')
-        self.certificate = Certificate(code='CRT')
-        self.department2.certificates['CRT'] = self.certificate
-        self.course = Course(code='CRS')
-        self.department2.certificates['CRT'].addCertCourse(self.course)
-        self.cert_course = self.certificate['CRS_100']
-        self.department2.courses['CRS'] = self.course
-        self.site3 = dict(faculties=dict(FAC=dict(DPT=self.department1)))
-        self.site4 = dict(faculties=dict(FAC=dict(DPT=self.department2)))
-        self.row = dict(
-            department_code='DPT',
-            faculty_code='FAC',
-            certificate_code='CRT',
-            course=self.course, level='100',
-            code='CRS_100')
+        self.certificate.addCertCourse(self.course)
+        self.row1 = dict(
+            department_code='DEP1',
+            faculty_code='FAC1',
+            certificate_code='CRT1',
+            course=self.course, level='100')
+        self.row2 = dict(
+            department_code='DEP1',
+            faculty_code='FAC1',
+            certificate_code='CRT2',
+            course=self.course, level='100')
+        self.csv_file_certificatecourse = os.path.join(
+            self.workdir, 'sample_certificatecourse_data.csv')
+        open(self.csv_file_certificatecourse, 'wb').write(CERTIFICATECOURSE_SAMPLE_DATA)
         return
 
@@ -672 +688 @@
     def test_parents_exist(self):
         # make sure we can find all certificate parents
-        result0 = self.proc.parentsExist(self.row, self.site0)
-        result1 = self.proc.parentsExist(self.row, self.site1)
-        result2 = self.proc.parentsExist(self.row, self.site2)
-        result3 = self.proc.parentsExist(self.row, self.site3)
-        result4 = self.proc.parentsExist(self.row, self.site4)
-        self.assertTrue(result0 is False)
-        self.assertTrue(result1 is False)
-        self.assertTrue(result2 is False)
-        self.assertTrue(result3 is False)
-        self.assertTrue(result4 is True)
+        result1 = self.proc.parentsExist(self.row1, self.app)
+        self.assertTrue(result1)
+        result2 = self.proc.parentsExist(self.row2, self.app)
+        self.assertFalse(result2)
         return
 
     def test_entry_exists(self):
         # make sure we find an entry if it exists
-        result0 = self.proc.entryExists(self.row, self.site0)
-        result1 = self.proc.entryExists(self.row, self.site1)
-        result2 = self.proc.entryExists(self.row, self.site2)
-        result3 = self.proc.entryExists(self.row, self.site3)
-        result4 = self.proc.entryExists(self.row, self.site4)
-        self.assertTrue(result0 is False)
-        self.assertTrue(result1 is False)
-        self.assertTrue(result2 is False)
-        self.assertTrue(result3 is False)
-        self.assertTrue(result4 is True)
+        result1 = self.proc.entryExists(self.row1, self.app)
+        self.assertTrue(result1)
+        result2 = self.proc.entryExists(self.row2, self.app)
+        self.assertFalse(result2)
         return
 
     def test_get_entry(self):
         # make sure we can get an entry if it exists
-        result1 = self.proc.getEntry(self.row, self.site3)
-        result2 = self.proc.getEntry(self.row, self.site4)
-        self.assertTrue(result1 is None)
-        self.assertTrue(result2 is self.cert_course)
+        result1 = self.proc.getEntry(self.row1, self.app)
+        self.assertTrue(result1 is self.certificate['CRS1_100'])
+        result2 = self.proc.getEntry(self.row2, self.app)
+        self.assertFalse(result2 is self.certificate['CRS1_100'])
         return
 
     def test_del_entry(self):
         # make sure we can delete entries
-        self.assertTrue('CRS_100' in self.certificate.keys())
-        self.proc.delEntry(self.row, self.site4)
-        self.assertTrue('CRS_100' not in self.certificate.keys())
-        return
+        self.assertTrue('CRS1_100' in self.certificate.keys())
+        self.proc.delEntry(self.row1, self.app)
+        self.assertTrue('CRS1_100' not in self.certificate.keys())
+        return
+
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.proc.doImport(
+            self.csv_file_certificatecourse, CERTIFICATECOURSE_HEADER_FIELDS)
+        content = open(fail_file).read()
+        self.assertEqual(num_warns,2)
+        self.assertEqual(
+            content,
+            'faculty_code,course,level,department_code,certificate_code,'
+            '--ERRORS--\r\nFAC1,CRS1,100,DEP1,CRT1,'
+            'This object already exists. Skipping.\r\nFAC1,CRS1,100,DEP1,CRT2,'
+            'Not all parents do exist yet. Skipping\r\n'
+
+            )
+        logcontent = open(self.logfile).read()
+        # Logging message from updateEntry
+        self.assertTrue(
+            'INFO - system - CertificateCourse Processor - '
+            'sample_certificatecourse_data - CRT1 - updated: '
+            'course=CRS1, level=200\n'
+            in logcontent)
+
+        return