Changeset 9960 for main/waeup.kofa/trunk/src/waeup/kofa/students

Timestamp:

16 Feb 2013, 12:20:06 (12 years ago)

Author:

Henrik Bettermann

Message:

Extend StudentStudyCourseProcessor? so that we can transfer students by import.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/students

Files:

• batching.py (modified) (7 diffs)
• interfaces.py (modified) (1 diff)
• tests/test_batching.py (modified) (5 diffs)

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

@@ -31 +31 @@
 from zope.interface import Interface
 from zope.schema import getFields
-from zope.component import queryUtility, getUtility
+from zope.component import queryUtility, getUtility, createObject
 from zope.event import notify
 from zope.catalog.interfaces import ICatalog
@@ -41 +41 @@
 from waeup.kofa.interfaces import MessageFactory as _
 from waeup.kofa.students.interfaces import (
-    IStudent, IStudentStudyCourse,
+    IStudent, IStudentStudyCourse, IStudentStudyCourseTransfer,
     IStudentUpdateByRegNo, IStudentUpdateByMatricNo,
     IStudentStudyLevel, ICourseTicketImport,
@@ -408 +408 @@
     name = u'StudentStudyCourse Processor (update only)'
     iface = IStudentStudyCourse
+    iface_transfer = IStudentStudyCourseTransfer
     factory_name = 'waeup.StudentStudyCourse'
 
@@ -425 +426 @@
         """Update obj to the values given in row.
         """
+        entry_mode = row.get('entry_mode', None)
+        certificate = row.get('certificate', None)
+        # If students are being transferred a new studycourse is created and
+        # the old one is moved.
+        if entry_mode == 'transfer':
+            studycourse = createObject(u'waeup.StudentStudyCourse')
+            student = obj.__parent__
+            studycourse.entry_session = obj.entry_session
+            if 'studycourse_1' in student.keys():
+                student['studycourse_2'] = obj
+            else:
+                student['studycourse_1'] = obj
+            del student['studycourse']
+            student['studycourse'] = studycourse
+            student.__parent__.logger.info(
+                '%s - transferred from %s to %s' % (
+                student.student_id, obj.certificate.code, certificate.code))
+            history = IObjectHistory(student)
+            history.addMessage('Transferred from %s to %s' % (
+                obj.certificate.code, certificate.code))
+            obj = student['studycourse']
         items_changed = super(StudentStudyCourseProcessor, self).updateEntry(
             obj, row, site, filename)
@@ -438 +460 @@
         """Validates all values in row.
         """
-        errs, inv_errs, conv_dict = super(
-            StudentStudyCourseProcessor, self).checkConversion(row, mode=mode)
+        # We have to use the correct interface. Transfer
+        # updates have different constraints.
+        entry_mode = row.get('entry_mode', None)
+        if entry_mode == 'transfer':
+            converter = IObjectConverter(self.iface_transfer)
+        else:
+            converter = IObjectConverter(self.iface)
+        errs, inv_errs, conv_dict =  converter.fromStringDict(
+            row, self.factory_name, mode=mode)
+
         # We have to check if current_level is in range of certificate.
         if 'certificate' in conv_dict and 'current_level' in conv_dict:
@@ -454 +484 @@
         problems.
         """
+        certificate = getattr(obj, 'certificate', None)
+        entry_session = getattr(obj, 'entry_session', None)
         current_level = row.get('current_level', None)
+        entry_mode = row.get('entry_mode', None)
+        # We have to ensure that the student can be transferred.
+        if entry_mode == 'transfer':
+            if certificate is None or entry_session is None:
+                return 'Former study course record incomplete.'
+            if 'studycourse_1' in obj.__parent__.keys() and \
+                'studycourse_2' in obj.__parent__.keys():
+                return 'Maximum number of transfers exceeded.'
         if current_level:
             if current_level == 999 and \
                 obj.__parent__.state in FORBIDDEN_POSTGRAD_STATES:
                 return 'Not a pg student.'
-            certificate = getattr(obj, 'certificate', None)
             cert = row.get('certificate', None)
             if certificate is None and cert is None:
                 return 'No certificate to check level.'
-            if certificate is not None and (
+            if certificate is not None and cert is None and (
                 current_level < certificate.start_level or \
                 current_level > certificate.end_level+120):
@@ -779 +818 @@
     factory_name = 'waeup.StudentStudyCourse'
 
-    additional_fields = [
-        'current_session',
-        'current_level',
-        'bypass_validation',
-        'validated_by']
-
     def checkUpdateRequirements(self, obj, row, site):
         """Checks requirements the studycourse and the student must fulfill

main/waeup.kofa/trunk/src/waeup/kofa/students/interfaces.py

@@ -460 +460 @@
         )
 
+    entry_mode = schema.Choice(
+        title = _(u'Entry Mode'),
+        source = StudyModeSource(),
+        required = False,
+        readonly = False,
+        )
+
+    entry_session = schema.Choice(
+        title = _(u'Entry Session'),
+        source = academic_sessions_vocab,
+        #default = datetime.now().year,
+        required = False,
+        readonly = False,
+        )
+
 
 IStudentStudyCourseTransfer['certificate'].order = IStudentStudyCourse[

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

@@ -27 +27 @@
 from time import time
 from zope.event import notify
-from zope.component import createObject
+from zope.component import createObject, queryUtility
 from zope.component.hooks import setSite, clearSite
+from zope.catalog.interfaces import ICatalog
 from zope.interface.verify import verifyClass, verifyObject
 from hurry.workflow.interfaces import IWorkflowState
@@ -88 +89 @@
 
 STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+TRANSFER_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_transfer_data.csv'),
+    'rb').read()
+
+TRANSFER_HEADER_FIELDS = TRANSFER_SAMPLE_DATA.split(
     '\n')[0].split(',')
 
@@ -571 +579 @@
             self.workdir, 'sample_studycourse_data.csv')
         open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
+        self.csv_file_transfer = os.path.join(
+            self.workdir, 'sample_transfer_data.csv')
+        open(self.csv_file_transfer, 'wb').write(TRANSFER_SAMPLE_DATA)
         return
 
@@ -639 +650 @@
             dict(reg_number='1', current_level=100), self.app)
         self.assertEqual(err, 'No certificate to check level.')
+        # When transferring students the method also checks
+        # if the former studycourse is complete.
+        err = self.processor.checkUpdateRequirements(
+            self.student['studycourse'],
+            dict(reg_number='1', certificate='CERT1', current_level=200,
+            entry_mode='transfer'), self.app)
+        self.assertEqual(err, 'Former study course record incomplete.')
+        self.student['studycourse'].certificate = self.certificate
+        self.student['studycourse'].entry_session = 2005
+        # The method doesn't care if current_level
+        # is not in range of CERT1. This is done by checkConversion
+        # if certificate is in row.
+        err = self.processor.checkUpdateRequirements(
+            self.student['studycourse'],
+            dict(reg_number='1', certificate='CERT1', current_level=200,
+            entry_mode='transfer'), self.app)
+        self.assertTrue(err is None)
 
     def test_import(self):
@@ -649 +677 @@
         self.assertEqual(studycourse.certificate.code, u'CERT1')
         shutil.rmtree(os.path.dirname(fin_file))
+
+    def test_import_transfer(self):
+        self.certificate2 = createObject('waeup.Certificate')
+        self.certificate2.code = 'CERT2'
+        self.certificate2.application_category = 'basic'
+        self.certificate2.start_level = 200
+        self.certificate2.end_level = 500
+        self.certificate2.study_mode = u'ug_pt'
+        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
+            self.certificate2)
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file_transfer, TRANSFER_HEADER_FIELDS,'update')
+        self.assertEqual(num_warns,0)
+        self.assertEqual(self.student['studycourse'].certificate.code, 'CERT2')
+        self.assertEqual(self.student['studycourse_1'].certificate.code, 'CERT1')
+        self.assertEqual(self.student['studycourse'].entry_mode, 'transfer')
+        self.assertEqual(self.student['studycourse_1'].entry_mode, 'ug_ft')
+        self.assertEqual(self.student.current_mode, 'ug_pt')
+        shutil.rmtree(os.path.dirname(fin_file))
+        # Transer has bee logged.
+        logcontent = open(self.logfile).read()
+        self.assertTrue(
+            'INFO - system - K1000000 - transferred from CERT1 to CERT2\n'
+            in logcontent)
+        self.assertTrue(
+            'INFO - system - '
+            'StudentStudyCourse Processor (update only) - '
+            'sample_transfer_data - K1000000 - updated: entry_mode=transfer, '
+            'certificate=CERT2, current_session=2009, current_level=300'
+            in logcontent)
+        # A history message has been added.
+        history = ' '.join(self.student.history.messages)
+        self.assertTrue(
+            "Transferred from CERT1 to CERT2 by system" in history)
+        # The catalog has been updated
+        cat = queryUtility(ICatalog, name='students_catalog')
+        results = list(
+            cat.searchResults(
+            certcode=('CERT2', 'CERT2')))
+        self.assertTrue(results[0] is self.student)
+        results = list(
+            cat.searchResults(
+            current_session=(2009, 2009)))
+        self.assertTrue(results[0] is self.student)
+        results = list(
+            cat.searchResults(
+            certcode=('CERT1', 'CERT1')))
+        self.assertEqual(len(results), 0)
 
 class StudentStudyLevelProcessorTest(StudentImportExportSetup):