Changeset 7548 for main/waeup.sirp/trunk/src/waeup/sirp/students

Timestamp:

1 Feb 2012, 11:19:56 (13 years ago)

Author:

Henrik Bettermann

Message:

Implement course ticket importer.

Add checkConversion tests.

Location:

main/waeup.sirp/trunk/src/waeup/sirp/students

Files:

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

main/waeup.sirp/trunk/src/waeup/sirp/students/batching.py

@@ -28 +28 @@
 from zope.interface import Interface
 from zope.schema import getFields
-from zope.component import queryUtility
+from zope.component import queryUtility, getUtility
 from zope.event import notify
 from zope.catalog.interfaces import ICatalog
@@ -38 +38 @@
     IStudent, IStudentStudyCourse,
     IStudentUpdateByRegNo, IStudentUpdateByMatricNo,
-    IStudentStudyLevel)
+    IStudentStudyLevel, ICourseTicket)
 from waeup.sirp.students.workflow import  IMPORTABLE_STATES
 from waeup.sirp.utils.batching import BatchProcessor
@@ -251 +251 @@
         elif 'reg_number' in row.keys() and row['reg_number']:
             reg_number = row['reg_number']
-            #import pdb; pdb.set_trace()
             cat = queryUtility(ICatalog, name='students_catalog')
             results = list(
@@ -296 +295 @@
             row, self.factory_name)
         # We have to check if current_level is in range of certificate.
-        # This is not done by the converter.
+        # This is not done by the converter. This kind of conversion
+        # checking does only work if a combination of certificate and
+        # current_level is provided.
         if conv_dict.has_key('certificate'):
           certificate = conv_dict['certificate']
@@ -355 +356 @@
         elif 'reg_number' in row.keys() and row['reg_number']:
             reg_number = row['reg_number']
-            #import pdb; pdb.set_trace()
             cat = queryUtility(ICatalog, name='students_catalog')
             results = list(
@@ -395 +395 @@
             row, self.factory_name)
         # We have to check if level is a valid integer.
-        # This is not by the converter.
+        # This is not done by the converter.
         try:
             level = int(row['level'])
@@ -403 +403 @@
             errs.append(('level','no integer'))
         return errs, inv_errs, conv_dict
+
+class CourseTicketProcessor(BatchProcessor):
+    """A batch processor for ICourseTicket objects.
+    """
+    grok.implements(IBatchProcessor)
+    grok.provides(IBatchProcessor)
+    grok.context(Interface)
+    util_name = 'courseticketimporter'
+    grok.name(util_name)
+
+    name = u'CourseTicket Importer'
+    iface = ICourseTicket
+    factory_name = 'waeup.CourseTicket'
+
+    location_fields = []
+
+    mode = None
+
+    @property
+    def available_fields(self):
+        return sorted(list(set(
+            ['student_id','reg_number','matric_number','level','code'] + getFields(
+                self.iface).keys())))
+
+    def checkHeaders(self, headerfields, mode='ignore'):
+        if not 'reg_number' in headerfields and not 'student_id' \
+            in headerfields and not 'matric_number' in headerfields:
+            raise FatalCSVError(
+                "Need at least columns student_id " +
+                "or reg_number or matric_number for import!")
+        if not 'level' in headerfields:
+            raise FatalCSVError(
+                "Need level for import!")
+        if not 'code' in headerfields:
+            raise FatalCSVError(
+                "Need code for import!")
+        # Check for fields to be ignored...
+        not_ignored_fields = [x for x in headerfields
+                              if not x.startswith('--')]
+        if len(set(not_ignored_fields)) < len(not_ignored_fields):
+            raise FatalCSVError(
+                "Double headers: each column name may only appear once.")
+        return True
+
+    def getParent(self, row, site):
+        if not 'students' in site.keys():
+            return None
+        if 'student_id' in row.keys() and row['student_id']:
+            if row['student_id'] in site['students']:
+                student = site['students'][row['student_id']]
+                return student['studycourse'].get(row['level'])
+        elif 'reg_number' in row.keys() and row['reg_number']:
+            reg_number = row['reg_number']
+            #import pdb; pdb.set_trace()
+            cat = queryUtility(ICatalog, name='students_catalog')
+            results = list(
+                cat.searchResults(reg_number=(reg_number, reg_number)))
+            if results:
+                return results[0]['studycourse'].get(row['level'])
+        elif 'matric_number' in row.keys() and row['matric_number']:
+            matric_number = row['matric_number']
+            cat = queryUtility(ICatalog, name='students_catalog')
+            results = list(
+                cat.searchResults(matric_number=(matric_number, matric_number)))
+            if results:
+                return results[0]['studycourse'].get(row['level'])
+        return None
+
+    def parentsExist(self, row, site):
+        return self.getParent(row, site) is not None
+
+    def entryExists(self, row, site):
+        return self.getEntry(row, site) is not None
+
+    def getEntry(self, row, site):
+        level = self.getParent(row, site)
+        if level is None:
+            return None
+        return level.get(row['code'])
+
+    def addEntry(self, obj, row, site):
+        parent = self.getParent(row, site)
+        catalog = getUtility(ICatalog, name='courses_catalog')
+        entries = list(catalog.searchResults(code=(row['code'],row['code'])))
+        obj.fcode = entries[0].__parent__.__parent__.__parent__.code
+        obj.dcode = entries[0].__parent__.__parent__.code
+        obj.title = entries[0].title
+        obj.credits = entries[0].credits
+        obj.passmark = entries[0].passmark
+        obj.semester = entries[0].semester
+        parent[row['code']] = obj
+        return
+
+    def checkConversion(self, row, mode='ignore'):
+        """Validates all values in row.
+        """
+        converter = IObjectConverter(self.iface)
+        errs, inv_errs, conv_dict =  converter.fromStringDict(
+            row, self.factory_name)
+        # We have to check if course really exists.
+        # This is not done by the converter.
+        catalog = getUtility(ICatalog, name='courses_catalog')
+        entries = catalog.searchResults(code=(row['code'],row['code']))
+        if len(entries) == 0:
+            errs.append(('code','non-existent'))
+            return errs, inv_errs, conv_dict
+        return errs, inv_errs, conv_dict

main/waeup.sirp/trunk/src/waeup/sirp/students/studylevel.py

@@ -103 +103 @@
 
 CourseTicket = attrs_to_fields(CourseTicket)
+
+class CourseTicketFactory(grok.GlobalUtility):
+    """A factory for student study levels.
+    """
+    grok.implements(IFactory)
+    grok.name(u'waeup.CourseTicket')
+    title = u"Create a new course ticket.",
+    description = u"This factory instantiates new course ticket instances."
+
+    def __call__(self, *args, **kw):
+        return CourseTicket()
+
+    def getInterfaces(self):
+        return implementedBy(CourseTicket)

main/waeup.sirp/trunk/src/waeup/sirp/students/tests/test_batching.py

@@ -32 +32 @@
 from waeup.sirp.students.batching import (
     StudentProcessor, StudentStudyCourseProcessor,
-    StudentStudyLevelProcessor)
+    StudentStudyLevelProcessor, CourseTicketProcessor)
 from waeup.sirp.students.student import Student
 from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
@@ -77 +77 @@
 
 STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+COURSETICKET_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
+    'rb').read()
+
+COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
     '\n')[0].split(',')
 
@@ -161 +168 @@
         assert len(self.app['students'].keys()) == 2
 
+    def test_checkConversion(self):
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', reg_state='admitted'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', reg_state=''))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('reg_state', 'no value provided') in errs)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', reg_state='nonsense'))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('reg_state', 'not allowed') in errs)
+
     def test_delEntry(self):
         assert self.student.student_id in self.app['students'].keys()
@@ -283 +303 @@
         self.assertEqual(student.reg_number,'1')
 
+    def test_checkConversion(self):
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', certificate='CERT1', current_level='200'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', certificate='CERT999'))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('certificate', u'Invalid value') in errs)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', certificate='CERT1', current_level='100'))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('current_level','not in range') in errs)
+        # If we import only current_level, no conversion checking is done.
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', current_level='100'))
+        self.assertEqual(len(errs),0)
+
     def test_import(self):
         num, num_warns, fin_file, fail_file = self.importer.doImport(
@@ -350 +387 @@
             IBatchProcessor, StudentStudyLevelProcessor) is True
 
+    def test_checkConversion(self):
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', level='220'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', level='900'))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('level','no valid integer') in errs)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', level='xyz'))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('level','no integer') in errs)
+
     def test_import(self):
         num, num_warns, fin_file, fail_file = self.importer.doImport(
@@ -365 +415 @@
         
 
+class CourseTicketImporterTest(FunctionalTestCase):
+
+    layer = FunctionalLayer
+
+    def setUp(self):
+        super(CourseTicketImporterTest, self).setUp()
+        self.dc_root = tempfile.mkdtemp()
+        self.workdir = tempfile.mkdtemp()
+        app = University()
+        app['datacenter'].setStoragePath(self.dc_root)
+        self.getRootFolder()['app'] = app
+        self.app = self.getRootFolder()['app']
+        setSite(app)
+
+        # Import students with subobjects
+        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
+        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
+        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
+            student_file, STUDENT_HEADER_FIELDS)
+        shutil.rmtree(os.path.dirname(fin_file))
+
+        # Populate university
+        self.certificate = createObject('waeup.Certificate')
+        self.certificate.code = 'CERT1'
+        self.certificate.application_category = 'basic'
+        self.certificate.start_level = 200
+        self.certificate.end_level = 500
+        self.app['faculties']['fac1'] = Faculty()
+        self.app['faculties']['fac1']['dep1'] = Department()
+        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
+            self.certificate)
+        self.course = createObject('waeup.Course')
+        self.course.code = 'COURSE1'
+        self.course.semester = 1
+        self.course.credits = 10
+        self.course.passmark = 40
+        self.app['faculties']['fac1']['dep1'].courses.addCourse(
+            self.course)
+        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
+            self.course, level=100)
+
+        # Update study courses
+        studycourse_file = os.path.join(
+            self.workdir, 'sample_studycourse_data.csv')
+        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
+        importer = StudentStudyCourseProcessor()
+        num, num_warns, fin_file, fail_file = importer.doImport(
+            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
+        shutil.rmtree(os.path.dirname(fin_file))
+
+        # Import study levels
+        importer = StudentStudyLevelProcessor()
+        studylevel_file = os.path.join(
+            self.workdir, 'sample_studylevel_data.csv')
+        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
+        num, num_warns, fin_file, fail_file = importer.doImport(
+            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
+        shutil.rmtree(os.path.dirname(fin_file))
+
+        self.importer = CourseTicketProcessor()
+        self.csv_file = os.path.join(
+            self.workdir, 'sample_courseticket_data.csv')
+        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
+
+    def tearDown(self):
+        super(CourseTicketImporterTest, self).tearDown()
+        shutil.rmtree(self.workdir)
+        shutil.rmtree(self.dc_root)
+        clearSite()
+        return
+
+    def test_interface(self):
+        # Make sure we fulfill the interface contracts.
+        assert verifyObject(IBatchProcessor, self.importer) is True
+        assert verifyClass(
+            IBatchProcessor, CourseTicketProcessor) is True
+
+    def test_checkConversion(self):
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', code='COURSE1', level='220'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', code='COURSE2', level='220'))
+        self.assertEqual(len(errs),1)
+        self.assertTrue(('code','non-existent') in errs)
+
+    def test_import(self):
+
+        num, num_warns, fin_file, fail_file = self.importer.doImport(
+            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
+
+        self.assertEqual(num_warns,2)
+        assert self.importer.entryExists(
+            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
+        courseticket = self.importer.getEntry(
+            dict(reg_number='1', level='100', code='COURSE1'), self.app)
+        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
+        self.assertEqual(courseticket.score, 1)
+        self.assertEqual(courseticket.core_or_elective, True)
+        self.assertEqual(courseticket.fcode, 'NA')
+        self.assertEqual(courseticket.dcode, 'NA')
+        self.assertEqual(courseticket.code, 'COURSE1')
+        self.assertEqual(courseticket.title, 'Unnamed Course')
+        self.assertEqual(courseticket.credits, 10)
+        self.assertEqual(courseticket.passmark, 40)
+        self.assertEqual(courseticket.semester, 1)
+        #import pdb; pdb.set_trace()
+        shutil.rmtree(os.path.dirname(fin_file))
+
+
 def test_suite():
     suite = unittest.TestSuite()
     for testcase in [
         StudentImporterTest,StudentStudyCourseImporterTest,
-        StudentStudyLevelImporterTest,]:
+        StudentStudyLevelImporterTest,CourseTicketImporterTest,]:
         suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
                 testcase