Ignore:
Timestamp:
10 Sep 2012, 11:05:07 (12 years ago)
Author:
uli
Message:

Merge changes from trunk, r8786-HEAD

Location:
main/waeup.kofa/branches/uli-async-update
Files:
1 deleted
14 edited
6 copied

Legend:

Unmodified
Added
Removed

  • main/waeup.kofa/branches/uli-async-update

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/api.txt

    r7811 r9169  
    2828   :maxdepth: 1
    2929
    30    courserefs.txt
     30   certcourses.txt

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/batching.py

    r8302 r9169  
    2929from zope.schema import getFields
    3030from zope.catalog.interfaces import ICatalog
    31 from waeup.kofa.interfaces import IBatchProcessor, IGNORE_MARKER, FatalCSVError
     31from zope.event import notify
     32from zope.securitypolicy.interfaces import (
     33    IPrincipalRoleManager, IPrincipalRoleMap)
     34from waeup.kofa.authentication import LocalRoleSetEvent
     35from waeup.kofa.interfaces import (
     36    IBatchProcessor, IGNORE_MARKER, DELETION_MARKER, FatalCSVError)
    3237from waeup.kofa.university.interfaces import (
    3338    IFacultiesContainer, IFaculty, ICourse, IDepartment, ICertificate,
    3439    ICertificateCourse)
     40from waeup.kofa.university import (
     41    Faculty, Department, Course, Certificate)
    3542from waeup.kofa.utils.batching import BatchProcessor
    3643
    … …  
    4653    name = u'Faculty Processor'
    4754    iface = IFaculty
     55    allowed_roles = Faculty.local_roles
    4856
    4957    location_fields = ['code',]
    … …  
    5462    def parentsExist(self, row, site):
    5563        return 'faculties' in site.keys()
     64
     65    @property
     66    def available_fields(self):
     67        fields = getFields(self.iface)
     68        return sorted(list(set(
     69            self.location_fields + fields.keys() + ['local_roles']
     70            )))
    5671
    5772    def entryExists(self, row, site):
    … …  
    7792        pass
    7893
    79 class DepartmentProcessor(BatchProcessor):
     94    def updateEntry(self, obj, row, site):
     95        """Update obj to the values given in row.
     96        """
     97        items_changed = ''
     98
     99        if row.has_key('local_roles') and row['local_roles'] not in (
     100            None, IGNORE_MARKER):
     101            role_manager = IPrincipalRoleManager(obj)
     102            role_map = IPrincipalRoleMap(obj)
     103            # Remove all existing local roles.
     104            for local_role, user_name, setting in role_map.getPrincipalsAndRoles():
     105                role_manager.unsetRoleForPrincipal(local_role, user_name)
     106                notify(LocalRoleSetEvent(
     107                        obj, local_role, user_name, granted=False))
     108            # Add new local roles.
     109            if row['local_roles'] != DELETION_MARKER:
     110                local_roles = eval(row['local_roles'])
     111                for rolemap in local_roles:
     112                    user = rolemap['user_name']
     113                    local_role = rolemap['local_role']
     114                    role_manager.assignRoleToPrincipal(local_role, user)
     115                    notify(LocalRoleSetEvent(obj, local_role, user, granted=True))
     116                    items_changed += (
     117                        '%s=%s, ' % ('local_roles', '%s|%s' % (user,local_role)))
     118            row.pop('local_roles')
     119
     120        # apply other values...
     121        items_changed += super(FacultyProcessor, self).updateEntry(
     122            obj, row, site)
     123
     124        # Log actions...
     125        location_field = self.location_fields[0]
     126        grok.getSite().logger.info('%s - %s - Record updated: %s'
     127            % (self.name, row[location_field], items_changed))
     128        return items_changed
     129
     130    def checkConversion(self, row, mode='create'):
     131        """Validates all values in row.
     132        """
     133        errs, inv_errs, conv_dict =  super(
     134            FacultyProcessor, self).checkConversion(row, mode=mode)
     135        if row.has_key('local_roles'):
     136            if row['local_roles'] in (None, DELETION_MARKER, IGNORE_MARKER):
     137                return errs, inv_errs, conv_dict
     138            try:
     139                local_roles = eval(row['local_roles'])
     140            except:
     141                errs.append(('local_roles','Error'))
     142                return errs, inv_errs, conv_dict
     143            if not isinstance(local_roles, list):
     144                errs.append(('local_roles','no list'))
     145                return errs, inv_errs, conv_dict
     146            for rolemap in local_roles:
     147                if not isinstance(rolemap, dict):
     148                    errs.append(('local_roles','no dicts'))
     149                    return errs, inv_errs, conv_dict
     150                if not 'user_name' in rolemap.keys() or not \
     151                    'local_role' in rolemap.keys():
     152                    errs.append(('local_roles','user_name or local_role missing'))
     153                    return errs, inv_errs, conv_dict
     154                local_role = rolemap['local_role']
     155                if not local_role in self.allowed_roles:
     156                    errs.append(('local_roles','%s not allowed' % local_role))
     157                    return errs, inv_errs, conv_dict
     158                user = rolemap['user_name']
     159                users = grok.getSite()['users']
     160                if not user in users.keys():
     161                    errs.append(('local_roles','%s does not exist' % user))
     162                    return errs, inv_errs, conv_dict
     163        return errs, inv_errs, conv_dict
     164
     165class DepartmentProcessor(FacultyProcessor):
    80166    """A batch processor for IDepartment objects.
    81167    """
    … …  
    88174    name = u'Department Processor'
    89175    iface = IDepartment
     176    allowed_roles = Department.local_roles
    90177
    91178    location_fields = ['code', 'faculty_code']
    … …  
    124211        return
    125212
    126 class CourseProcessor(BatchProcessor):
     213class CourseProcessor(FacultyProcessor):
    127214    """A batch processor for ICourse objects.
    128215    """
    … …  
    135222    name = u'Course Processor'
    136223    iface = ICourse
     224    allowed_roles = Course.local_roles
    137225
    138226    location_fields = ['code', 'faculty_code', 'department_code']
    … …  
    175263        return
    176264
    177 class CertificateProcessor(BatchProcessor):
     265class CertificateProcessor(FacultyProcessor):
    178266    """A batch processor for ICertificate objects.
    179267    """
    … …  
    186274    name = u'Certificate Processor'
    187275    iface = ICertificate
     276    allowed_roles = Certificate.local_roles
    188277
    189278    location_fields = ['code']
    … …  
    196285        fields = getFields(self.iface)
    197286        return sorted(list(set(
    198             ['faculty_code','department_code'] + fields.keys())))
     287            ['faculty_code','department_code'] + fields.keys()
     288            + ['local_roles'])))
    199289
    200290    def checkHeaders(self, headerfields, mode='create'):
    … …  
    269359        return
    270360
    271 class CertificateCourseProcessor(BatchProcessor):
     361class CertificateCourseProcessor(FacultyProcessor):
    272362    """A batch processor for ICertificateCourse objects.
    273363    """
    … …  
    317407    def addEntry(self, obj, row, site):
    318408        parent = self.getParent(row, site)
    319         parent.addCourseRef(row['course'],
     409        parent.addCertCourse(row['course'],
    320410                            row['level'], row['mandatory'])
    321411        return
    … …  
    323413    def delEntry(self, row, site):
    324414        parent = self.getParent(row, site)
    325         parent.delCourseRef(row['course'].code, row['level'])
    326         return
     415        parent.delCertCourse(row['course'].code, row['level'])
     416        return

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/catalog.py

    r7811 r9169  
    159159        self.title = context.course.title
    160160        self.code = context.getCourseCode
    161         self.type = 'Course Referrer'
     161        self.type = 'Certificate Course'
    162162
    163163def search(query=None, view=None):

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/certificate.py

    r8738 r9169  
    3636    grok.implements(ICertificate, ICertificateAdd)
    3737
    38     @property       # Make this method read_only and looking like an attr.
    39     def local_roles(self):
    40         return ['waeup.local.CourseAdviser100',
    41                 'waeup.local.CourseAdviser200',
    42                 'waeup.local.CourseAdviser300',
    43                 'waeup.local.CourseAdviser400',
    44                 'waeup.local.CourseAdviser500',
    45                 'waeup.local.CourseAdviser600',
    46                 ]
     38    local_roles = [
     39        'waeup.local.CourseAdviser100',
     40        'waeup.local.CourseAdviser200',
     41        'waeup.local.CourseAdviser300',
     42        'waeup.local.CourseAdviser400',
     43        'waeup.local.CourseAdviser500',
     44        'waeup.local.CourseAdviser600',
     45        ]
    4746
    4847    def __init__(self, code=u'NA', title=u'Unnamed Certificate',
    4948                 study_mode=None, start_level=None,
    5049                 end_level=None, application_category=None,
    51                  school_fee_1=None, school_fee_2=None):
     50                 school_fee_1=None, school_fee_2=None,
     51                 school_fee_3=None, school_fee_4=None):
    5252        super(Certificate, self).__init__()
    5353        self.code = code
    … …  
    5959        self.school_fee_1 = school_fee_1
    6060        self.school_fee_2 = school_fee_2
     61        self.school_fee_3 = school_fee_3
     62        self.school_fee_4 = school_fee_4
    6163
    6264    def longtitle(self):
    6365        return "%s (%s)" % (self.title,self.code)
    6466
    65     def addCourseRef(self, course, level=100, mandatory=True):
    66         """Add a course referrer.
     67    def addCertCourse(self, course, level=100, mandatory=True):
     68        """Add a certificate course.
    6769        """
    6870        code = "%s_%s" % (course.code, level)
    … …  
    7274        self._p_changed = True
    7375
    74     def delCourseRef(self, code, level=None):
    75         """Delete a course referrer denoted by its code.
     76    def delCertCourse(self, code, level=None):
     77        """Delete a certificate course denoted by its code.
    7678        """
    7779        keys = list(self.keys()) # create list copy

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/certificate.txt

    r7819 r9169  
    9696      .. XXX: This is not a proper description
    9797
    98    .. method:: addCourseRef(course[, level=100,[ mandatory=True]])
    99 
    100       Add a referrer of a course. A course is an object implementing
     98   .. method:: addCertCourse(course[, level=100,[ mandatory=True]])
     99
     100      Add a certificate course. A course is an object implementing
    101101      :class:`waeup.kofa.interfaces.ICourse`.
    102102
    103       Please don't be confused by the term 'referrer'. As course objects
    104       will normaly be  stored
    105       in a department, the course referrer refers to
     103      Please don't be confused by the term 'certificate course'.
     104      As course objects will normaly be  stored
     105      in a department, the certificate course refers to
    106106      the 'real' course in the department container.
    107107
    108    .. method:: delCourseRef(code)
    109 
    110       Remove a course referrer from a certificate.
    111 
    112       The referrer must be given by its code number.
     108   .. method:: delCertCourse(code)
     109
     110      Remove a certificate course from a certificate.
     111
     112      The certificate course must be given by its code number.
    113113
    114114:class:`CertificateCourse`
    … …  
    117117.. class:: CertificateCourse(course[, level=100[, mandatory=True]])
    118118
    119    Create a course referrer.
    120 
    121    A certificate course (old term) or course referrer are held in
     119   Create a certificate course.
     120
     121   A certificate course are held in
    122122   certificates and refer to an existing :class:`Course` instance held
    123123   elsewhere.
    124124
    125    A certificate can require several course referrers and one
    126    course referrer can be required by several certificates.
     125   A certificate can require several certificate courses and one
     126   certificate course can be required by several certificates.
    127127
    128128   .. attribute:: course
    … …  
    225225
    226226   To accomplish that, the parents of the removed course are looked up
    227    for a certifcate container which contains a course referrer that
     227   for a certifcate container which contains a certificate course that
    228228   refers to the deleted course.
    229229
    … …  
    341341    >>> mycert.code = 'MYCERT'
    342342    >>> mydept.certificates.addCertificate(mycert)
    343     >>> mycert.addCourseRef(mycourse)
     343    >>> mycert.addCertCourse(mycourse)
    344344
    345345    >>> transaction.commit()
    … …  
    354354    ['mydept']
    355355
    356 The course referrer we stored in the certificate is indeed a
     356The certificate course we stored in the certificate is indeed a
    357357referrer of the course, not a copy of it:
    358358

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/course.py

    r7811 r9169  
    2929    """
    3030    grok.implements(ICourse, ICourseAdd)
     31
     32    local_roles = ['waeup.local.Lecturer']
    3133
    3234    def __init__(self,
    … …  
    7880        # Remove that referrer...
    7981        cert = certcourse.__parent__
    80         cert.delCourseRef(code)
     82        cert.delCertCourse(code)
    8183        cert._p_changed = True
    8284    return

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/department.py

    r7841 r9169  
    3232    grok.implements(IDepartment, IDepartmentAdd)
    3333
    34     @property       # Make this method read_only and looking like an attr.
    35     def local_roles(self):
    36         return ['waeup.local.DepartmentManager',
    37                 'waeup.local.ClearanceOfficer',
    38                 'waeup.local.CourseAdviser100',
    39                 'waeup.local.CourseAdviser200',
    40                 'waeup.local.CourseAdviser300',
    41                 'waeup.local.CourseAdviser400',
    42                 'waeup.local.CourseAdviser500',
    43                 'waeup.local.CourseAdviser600',
    44                 ]
     34    local_roles = [
     35        'waeup.local.DepartmentManager',
     36        'waeup.local.ClearanceOfficer',
     37        'waeup.local.UGClearanceOfficer',
     38        'waeup.local.PGClearanceOfficer',
     39        'waeup.local.CourseAdviser100',
     40        'waeup.local.CourseAdviser200',
     41        'waeup.local.CourseAdviser300',
     42        'waeup.local.CourseAdviser400',
     43        'waeup.local.CourseAdviser500',
     44        'waeup.local.CourseAdviser600',
     45        ]
    4546
    4647    # A simple counter for ids.

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/export.py

    r8301 r9169  
    1919"""
    2020import grok
     21from zope.securitypolicy.interfaces import IPrincipalRoleMap
    2122from waeup.kofa.interfaces import ICSVExporter
    2223from waeup.kofa.interfaces import MessageFactory as _
    … …  
    3031
    3132    #: Fieldnames considered by this exporter
    32     fields = ('code', 'title', 'title_prefix')
     33    fields = ('code', 'title', 'title_prefix', 'users_with_local_roles')
    3334
    3435    #: The title under which this exporter will be displayed
    3536    title = _(u'Faculties')
     37
     38    def mangle_value(self, value, name, context=None):
     39        """Hook for mangling values in derived classes
     40        """
     41        if name == 'users_with_local_roles':
     42            value = []
     43            role_map = IPrincipalRoleMap(context)
     44            for local_role, user_name, setting in role_map.getPrincipalsAndRoles():
     45                value.append({'user_name':user_name,'local_role':local_role})
     46        return super(FacultyExporter, self).mangle_value(
     47            value, name, context)
    3648
    3749    def export(self, faculties, filepath=None):
    … …  
    5062        If `filepath` is ``None``, a raw string with CSV data is returned.
    5163        """
    52         writer, outfile = self.get_csv_writer(filepath)
    5364        faculties = site.get('faculties', {})
    5465        return self.export(faculties.values(), filepath)
    … …  
    6172
    6273    #: Fieldnames considered by this exporter
    63     fields = ('code', 'faculty_code', 'title', 'title_prefix')
     74    fields = ('code', 'faculty_code', 'title', 'title_prefix',
     75        'users_with_local_roles')
    6476
    6577    #: The title under which this exporter will be displayed
    … …  
    7789
    7890    def export_all(self, site, filepath=None):
    79         """Export faculties in facultycontainer into filepath as CSV data.
     91        """Export departments in faculty into filepath as CSV data.
    8092
    8193        If `filepath` is ``None``, a raw string with CSV data is returned.
    … …  
    97109    #: Fieldnames considered by this exporter
    98110    fields = ('code', 'faculty_code', 'department_code', 'title', 'credits',
    99               'passmark', 'semester')
     111              'passmark', 'semester', 'users_with_local_roles')
    100112
    101113    #: The title under which this exporter will be displayed
    … …  
    105117        """Hook for mangling values in derived classes
    106118        """
    107         if name == 'faculty_code':
     119        if name == 'users_with_local_roles':
     120            value = []
     121            role_map = IPrincipalRoleMap(context)
     122            for local_role, user_name, setting in role_map.getPrincipalsAndRoles():
     123                value.append({'user_name':user_name,'local_role':local_role})
     124        elif name == 'faculty_code':
    108125            try:
    109126                value = context.__parent__.__parent__.__parent__.code
    … …  
    119136
    120137    def export_all(self, site, filepath=None):
    121         """Export faculties in facultycontainer into filepath as CSV data.
     138        """Export courses into filepath as CSV data.
    122139
    123140        If `filepath` is ``None``, a raw string with CSV data is returned.
    … …  
    140157    fields = ('code', 'faculty_code', 'department_code', 'title', 'study_mode',
    141158              'start_level', 'end_level', 'application_category',
    142               'school_fee_1', 'school_fee_2')
     159              'school_fee_1', 'school_fee_2', 'school_fee_3', 'school_fee_4',
     160              'users_with_local_roles')
    143161
    144162    #: The title under which this exporter will be displayed
    … …  
    146164
    147165    def export_all(self, site, filepath=None):
    148         """Export faculties in facultycontainer into filepath as CSV data.
     166        """Export certificates into filepath as CSV data.
    149167
    150168        If `filepath` is ``None``, a raw string with CSV data is returned.
    … …  
    193211
    194212    def export_all(self, site, filepath=None):
    195         """Export faculties in facultycontainer into filepath as CSV data.
     213        """Export certificate courses into filepath as CSV data.
    196214
    197215        If `filepath` is ``None``, a raw string with CSV data is returned.

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/faculty.py

    r7841 r9169  
    3232    grok.implements(IFaculty, IFacultyAdd)
    3333
    34     @property       # Make this method read_only and looking like an attr.
    35     def local_roles(self):
    36         return ['waeup.local.DepartmentManager',
    37                 'waeup.local.ClearanceOfficer',
    38                 'waeup.local.CourseAdviser100',
    39                 'waeup.local.CourseAdviser200',
    40                 'waeup.local.CourseAdviser300',
    41                 'waeup.local.CourseAdviser400',
    42                 'waeup.local.CourseAdviser500',
    43                 'waeup.local.CourseAdviser600',
    44                 ]
     34    local_roles = [
     35        'waeup.local.DepartmentManager',
     36        'waeup.local.ClearanceOfficer',
     37        'waeup.local.UGClearanceOfficer',
     38        'waeup.local.PGClearanceOfficer',
     39        'waeup.local.CourseAdviser100',
     40        'waeup.local.CourseAdviser200',
     41        'waeup.local.CourseAdviser300',
     42        'waeup.local.CourseAdviser400',
     43        'waeup.local.CourseAdviser500',
     44        'waeup.local.CourseAdviser600',
     45        ]
    4546
    4647    def __init__(self,

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/interfaces.py

    r8472 r9169  
    240240        )
    241241
     242    school_fee_3 = schema.Float(
     243        title = _(u'Foreigner Initial School Fee'),
     244        required = False,
     245        )
     246
     247    school_fee_4 = schema.Float(
     248        title = _(u'Foreigner Returning School Fee'),
     249        required = False,
     250        )
     251
    242252    def longtitle():
    243253        """
    … …  
    286296    """
    287297    course = schema.Choice(
    288         title = _(u'Course Referrer'),
     298        title = _(u'Course'),
    289299        source = CourseSource(),
    290300        readonly = True,
    … …  
    300310    mandatory = schema.Bool(
    301311        title = _(u'Is mandatory course (not elective)'),
    302         required = True,
     312        required = False,
    303313        default = True,
    304314        )

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/test_batching.py

    r7811 r9169  
    1919# Tests for university related batching
    2020import unittest
    21 
     21import tempfile
     22import shutil
     23import os
     24from zope.component.hooks import setSite, clearSite
     25from zope.component import createObject
     26from zope.securitypolicy.interfaces import (
     27    IPrincipalRoleMap, IPrincipalRoleManager)
     28from zope.testbrowser.testing import Browser
    2229from zope.interface.verify import verifyClass, verifyObject
    2330from waeup.kofa.interfaces import IBatchProcessor
    2431from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
     32from waeup.kofa.app import University
    2533from waeup.kofa.university.batching import (
    2634    FacultyProcessor, DepartmentProcessor, CourseProcessor,
    … …  
    2836from waeup.kofa.university.certificate import Certificate, CertificateCourse
    2937from waeup.kofa.university.course import Course
    30 from waeup.kofa.university.department import Department
    31 
    32 
    33 class TestFacultyProcessor(unittest.TestCase):
     38from waeup.kofa.university import Faculty, Department
     39from waeup.kofa.university.batching import FacultyProcessor
     40
     41FACULTY_SAMPLE_DATA = open(
     42    os.path.join(os.path.dirname(__file__), 'sample_faculty_data.csv'),
     43    'rb').read()
     44
     45FACULTY_HEADER_FIELDS = FACULTY_SAMPLE_DATA.split(
     46    '\n')[0].split(',')
     47
     48FACULTY_SAMPLE_DATA_UPDATE = open(
     49    os.path.join(os.path.dirname(__file__), 'sample_faculty_data_update.csv'),
     50    'rb').read()
     51
     52FACULTY_HEADER_FIELDS_UPDATE = FACULTY_SAMPLE_DATA_UPDATE.split(
     53    '\n')[0].split(',')
     54
     55DEPARTMENT_SAMPLE_DATA = open(
     56    os.path.join(os.path.dirname(__file__), 'sample_department_data.csv'),
     57    'rb').read()
     58
     59DEPARTMENT_HEADER_FIELDS = DEPARTMENT_SAMPLE_DATA.split(
     60    '\n')[0].split(',')
     61
     62CERTIFICATE_SAMPLE_DATA = open(
     63    os.path.join(os.path.dirname(__file__), 'sample_certificate_data.csv'),
     64    'rb').read()
     65
     66CERTIFICATE_HEADER_FIELDS = CERTIFICATE_SAMPLE_DATA.split(
     67    '\n')[0].split(',')
     68
     69COURSE_SAMPLE_DATA = open(
     70    os.path.join(os.path.dirname(__file__), 'sample_course_data.csv'),
     71    'rb').read()
     72
     73COURSE_HEADER_FIELDS = COURSE_SAMPLE_DATA.split(
     74    '\n')[0].split(',')
     75
     76class UniversityProcessorSetup(FunctionalTestCase):
     77
     78    layer = FunctionalLayer
    3479
    3580    def setUp(self):
     81        super(UniversityProcessorSetup, self).setUp()
     82        self.dc_root = tempfile.mkdtemp()
     83        self.workdir = tempfile.mkdtemp()
     84        app = University()
     85        self.getRootFolder()['app'] = app
     86        self.app = self.getRootFolder()['app']
     87        setSite(app)
     88
     89        self.app['users'].addUser('bob', 'bobssecret')
     90        self.app['users'].addUser('anne', 'annessecret')
     91
     92        # Populate university
     93        self.certificate = createObject('waeup.Certificate')
     94        self.certificate.code = 'CRT1'
     95        self.app['faculties']['FAC1'] = Faculty(code='FAC1')
     96        self.app['faculties']['FAC1']['DEP1'] = Department(code='DEP1')
     97        self.app['faculties']['FAC1']['DEP1'].certificates.addCertificate(
     98            self.certificate)
     99        self.course = createObject('waeup.Course')
     100        self.course.code = 'CRS1'
     101        self.app['faculties']['FAC1']['DEP1'].courses.addCourse(
     102            self.course)
     103        #self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
     104        #    self.course, level=100)
     105       
     106        self.logfile = os.path.join(
     107            self.app['datacenter'].storage, 'logs', 'main.log')
     108        return
     109        return
     110
     111    def tearDown(self):
     112        super(UniversityProcessorSetup, self).tearDown()
     113        shutil.rmtree(self.workdir)
     114        shutil.rmtree(self.dc_root)
     115        clearSite()
     116        return
     117
     118class TestFacultyProcessor(UniversityProcessorSetup):
     119
     120    def setUp(self):
     121        super(TestFacultyProcessor, self).setUp()
     122
     123        self.browser = Browser()
     124        self.browser.handleErrors = False
     125        self.datacenter_path = 'http://localhost/app/datacenter'
     126        self.app['datacenter'].setStoragePath(self.dc_root)
     127
    36128        self.proc = FacultyProcessor()
    37129        self.site1 = dict(faculties=dict())
    38130        self.site2 = dict(faculties=dict(FAC='pseudo faculty'))
    39131        self.row = dict(code='FAC')
     132
     133        self.csv_file_faculty = os.path.join(self.workdir, 'sample_faculty_data.csv')
     134        open(self.csv_file_faculty, 'wb').write(FACULTY_SAMPLE_DATA)
     135        self.csv_file_faculty_update = os.path.join(self.workdir, 'sample_faculty_data_update.csv')
     136        open(self.csv_file_faculty_update, 'wb').write(FACULTY_SAMPLE_DATA_UPDATE)
    40137        return
    41138
    … …  
    60157        return
    61158
    62 class TestDepartmentProcessors(unittest.TestCase):
     159    def test_checkConversion(self):
     160        # Make sure we can check conversions.
     161        errs, inv_errs, conv_dict = self.proc.checkConversion(
     162            dict(faculty_code='ABC', local_roles='[]'))
     163        self.assertEqual(len(errs),0)
     164
     165        errs, inv_errs, conv_dict = self.proc.checkConversion(
     166            dict(faculty_code='ABC',
     167            local_roles="['nonsense'"
     168            ))
     169        self.assertEqual(len(errs),1)
     170        self.assertEqual(errs, [('local_roles', 'Error')])
     171        errs, inv_errs, conv_dict = self.proc.checkConversion(
     172            dict(faculty_code='ABC',
     173            local_roles="('abc')"
     174            ))
     175        self.assertEqual(len(errs),1)
     176        self.assertEqual(errs, [('local_roles', 'no list')])
     177        errs, inv_errs, conv_dict = self.proc.checkConversion(
     178            dict(faculty_code='ABC',
     179            local_roles="[('ABC')]"
     180            ))
     181        self.assertEqual(len(errs),1)
     182        self.assertEqual(errs, [('local_roles', 'no dicts')])
     183        errs, inv_errs, conv_dict = self.proc.checkConversion(
     184            dict(faculty_code='ABC',
     185            local_roles="('abc')"
     186            ))
     187        self.assertEqual(len(errs),1)
     188        self.assertEqual(errs, [('local_roles', 'no list')])
     189        errs, inv_errs, conv_dict = self.proc.checkConversion(
     190            dict(faculty_code='ABC', local_roles=
     191            "[{'name':'bob','local_role':'waeup.local.DepartmentManager'},]"
     192            ))
     193        self.assertEqual(len(errs),1)
     194        self.assertEqual(errs, [('local_roles', 'user_name or local_role missing')])
     195        errs, inv_errs, conv_dict = self.proc.checkConversion(
     196            dict(faculty_code='ABC', local_roles=
     197            "[{'user_name':'bob','localrole':'waeup.local.DepartmentManager'},]"
     198            ))
     199        self.assertEqual(len(errs),1)
     200        self.assertEqual(errs, [('local_roles', 'user_name or local_role missing')])
     201        errs, inv_errs, conv_dict = self.proc.checkConversion(
     202            dict(faculty_code='ABC', local_roles=
     203            "[{'user_name':'bob','local_role':'waeup.local.Boss'},]"
     204            ))
     205        self.assertEqual(len(errs),1)
     206        self.assertEqual(errs, [('local_roles', 'waeup.local.Boss not allowed')])
     207        errs, inv_errs, conv_dict = self.proc.checkConversion(
     208            dict(faculty_code='ABC', local_roles=
     209            "[{'user_name':'john','local_role':'waeup.local.DepartmentManager'},]"
     210            ))
     211        self.assertEqual(len(errs),1)
     212        self.assertEqual(errs, [('local_roles', 'john does not exist')])
     213        errs, inv_errs, conv_dict = self.proc.checkConversion(
     214            dict(faculty_code='ABC', local_roles=
     215            "[{'user_name':'bob','local_role':'waeup.local.DepartmentManager'},]"
     216            ))
     217        self.assertEqual(len(errs),0)
     218        return
     219
     220    def test_import(self):
     221        num, num_warns, fin_file, fail_file = self.proc.doImport(
     222            self.csv_file_faculty, FACULTY_HEADER_FIELDS)
     223        content = open(fail_file).read()
     224        self.assertEqual(num_warns,5)
     225        self.assertEqual(
     226            content,
     227            'code,local_roles,--ERRORS--\r\n'
     228            'CDE,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
     229            'local_roles: alice does not exist\r\n'
     230            'DEF,"[{\'user_name\':\'bob\',\'local_role\':\'waeup.local.Boss\'}]",'
     231            'local_roles: waeup.local.Boss not allowed\r\n'
     232            'EFG,[(\'anything\')],local_roles: no dicts\r\n'
     233            'FGH,[,local_roles: Error\r\n'
     234            'GHI,"[{\'user\':\'bob\',\'local\':\'waeup.local.DepartmentManager\'}]",'
     235            'local_roles: user_name or local_role missing\r\n'
     236            )
     237        # Bob got a local role in faculty ABC.
     238        abc = self.app['faculties']['ABC']
     239        role_map = IPrincipalRoleMap(abc)
     240        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
     241        self.assertEqual(user_name, 'bob')
     242        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
     243        shutil.rmtree(os.path.dirname(fin_file))
     244        return
     245
     246    def test_import_update(self):
     247        self.app['faculties']['FAC2'] = Faculty(code='FAC2')
     248        self.app['faculties']['FAC3'] = Faculty(code='FAC3')
     249        self.app['faculties']['FAC4'] = Faculty(code='FAC4')
     250
     251        role_manager1 = IPrincipalRoleManager(self.app['faculties']['FAC1'])
     252        role_manager1.assignRoleToPrincipal('alfonsrole', 'alfons')
     253        role_map1 = IPrincipalRoleMap(self.app['faculties']['FAC1'])
     254        self.assertEqual(len(role_map1.getPrincipalsAndRoles()), 1)
     255
     256        role_manager2 = IPrincipalRoleManager(self.app['faculties']['FAC2'])
     257        role_manager2.assignRoleToPrincipal('alfonsrole', 'alfons')
     258        role_map2 = IPrincipalRoleMap(self.app['faculties']['FAC2'])
     259        self.assertEqual(len(role_map2.getPrincipalsAndRoles()), 1)
     260
     261        role_manager3 = IPrincipalRoleManager(self.app['faculties']['FAC3'])
     262        role_manager3.assignRoleToPrincipal('alfonsrole', 'alfons')
     263        role_map3 = IPrincipalRoleMap(self.app['faculties']['FAC3'])
     264        self.assertEqual(len(role_map3.getPrincipalsAndRoles()), 1)
     265
     266        role_manager4 = IPrincipalRoleManager(self.app['faculties']['FAC4'])
     267        role_manager4.assignRoleToPrincipal('alfonsrole', 'alfons')
     268        role_map4 = IPrincipalRoleMap(self.app['faculties']['FAC4'])
     269        self.assertEqual(len(role_map4.getPrincipalsAndRoles()), 1)
     270
     271        local_role, user_name, setting = role_map2.getPrincipalsAndRoles()[0]
     272        self.assertEqual(user_name, 'alfons')
     273        self.assertEqual(local_role, 'alfonsrole')
     274
     275        num, num_warns, fin_file, fail_file = self.proc.doImport(
     276            self.csv_file_faculty_update, FACULTY_HEADER_FIELDS_UPDATE, 'update')
     277        self.assertEqual(num_warns,0)
     278        # Local roles have been removed in FAC1 due to deletion marker.
     279        self.assertEqual(len(role_map1.getPrincipalsAndRoles()), 0)
     280        # Old local roles have been removed and new roles have been added in FAC2.
     281        self.assertEqual(len(role_map2.getPrincipalsAndRoles()), 1)
     282        local_role, user_name, setting = role_map2.getPrincipalsAndRoles()[0]
     283        self.assertEqual(user_name, 'bob')
     284        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
     285        # Local roles are not touched in FAC3 due to ignore marker.
     286        self.assertEqual(len(role_map3.getPrincipalsAndRoles()), 1)
     287        local_role, user_name, setting = role_map3.getPrincipalsAndRoles()[0]
     288        self.assertEqual(user_name, 'alfons')
     289        self.assertEqual(local_role, 'alfonsrole')
     290        # Local roles are not touched in FAC4 due to empty cell.
     291        self.assertEqual(len(role_map4.getPrincipalsAndRoles()), 1)
     292        local_role, user_name, setting = role_map4.getPrincipalsAndRoles()[0]
     293        self.assertEqual(user_name, 'alfons')
     294        self.assertEqual(local_role, 'alfonsrole')
     295        shutil.rmtree(os.path.dirname(fin_file))
     296        return
     297       
     298    def test_import_update_logging(self):
     299        self.app['faculties']['FAC2'] = Faculty(code='FAC2')
     300        self.app['faculties']['FAC3'] = Faculty(code='FAC3')
     301        self.app['faculties']['FAC4'] = Faculty(code='FAC4')
     302        num, num_warns, fin_file, fail_file = self.proc.doImport(
     303            self.csv_file_faculty_update, FACULTY_HEADER_FIELDS_UPDATE, 'update')
     304        self.assertEqual(num_warns,0)
     305        logcontent = open(self.logfile).read()
     306        # Logging message from updateEntry
     307        self.assertTrue(
     308            'INFO - system - Department Processor - DEP2 - '
     309            'Record updated: local_roles=anne|waeup.local.DepartmentManager, '
     310            'code=DEP2'
     311            in logcontent)
     312
     313    def test_upload_import_reupload(self):
     314        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
     315        self.browser.open(self.datacenter_path)
     316        self.assertEqual(self.browser.headers['Status'], '200 Ok')
     317        self.assertEqual(self.browser.url, self.datacenter_path)
     318        self.browser.getLink("Upload data").click()
     319        file = open(self.csv_file_faculty)
     320        ctrl = self.browser.getControl(name='uploadfile:file')
     321        file_ctrl = ctrl.mech_control
     322        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
     323        self.browser.getControl('Upload').click()
     324        self.browser.getLink('Process data').click()
     325        self.browser.getControl(name="select").click()
     326        importerselect = self.browser.getControl(name='importer')
     327        importerselect.getControl('Faculty Processor').selected = True
     328        modeselect = self.browser.getControl(name='mode')
     329        modeselect.getControl(value='create').selected = True
     330        self.browser.getControl('Proceed to step 3').click()
     331        self.assertTrue('Header fields OK' in self.browser.contents)
     332        self.browser.getControl('Perform import').click()
     333        self.assertTrue('Successfully processed 1 rows' in self.browser.contents)
     334        # We import the same file a second time.
     335        self.browser.open(self.datacenter_path)
     336        self.browser.getLink("Upload data").click()
     337        file = open(self.csv_file_faculty)
     338        ctrl = self.browser.getControl(name='uploadfile:file')
     339        file_ctrl = ctrl.mech_control
     340        file_ctrl.add_file(file, filename='sample_faculty_data.csv')
     341        self.browser.getControl('Upload').click()
     342        self.assertTrue(
     343            'File with same name was uploaded earlier' in self.browser.contents)
     344        return
     345
     346class TestDepartmentProcessor(UniversityProcessorSetup):
    63347
    64348    def setUp(self):
     349        super(TestDepartmentProcessor, self).setUp()
    65350        self.proc = DepartmentProcessor()
    66351        self.site0 = dict()
    … …  
    69354        self.site3 = dict(faculties=dict(FAC=dict(DPT='pseudo department')))
    70355        self.row = dict(code='DPT', faculty_code='FAC')
     356
     357        self.csv_file_department = os.path.join(self.workdir, 'sample_department_data.csv')
     358        open(self.csv_file_department, 'wb').write(DEPARTMENT_SAMPLE_DATA)
    71359        return
    72360
    … …  
    115403        return
    116404
    117 class CourseProcessorTests(FunctionalTestCase):
    118 
    119     layer = FunctionalLayer
     405    def test_import(self):
     406        num, num_warns, fin_file, fail_file = self.proc.doImport(
     407            self.csv_file_department, DEPARTMENT_HEADER_FIELDS)
     408        content = open(fail_file).read()
     409        self.assertEqual(num_warns,6)
     410        self.assertEqual(
     411            content,
     412            'faculty_code,code,local_roles,--ERRORS--\r\n'
     413            'FAC1,DEP2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
     414            'local_roles: alice does not exist\r\n'
     415            'FAC1,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
     416            'local_roles: waeup.local.Boss not allowed\r\n'
     417            'FAC1,DEP2,[(\'anything\')],local_roles: no dicts\r\n'
     418            'FAC1,DEP2,[,local_roles: Error\r\n'
     419            'FAC1,DEP2,"[{\'user\':\'anne\',\'local\':\'waeup.local.DepartmentManager\'}]",'
     420            'local_roles: user_name or local_role missing\r\n'
     421            'FAC11,DEP2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.DepartmentManager\'}]",'
     422            'Not all parents do exist yet. Skipping\r\n'
     423            )
     424        # Anne got a local role in department DEP2.
     425        dep = self.app['faculties']['FAC1']['DEP2']
     426        role_map = IPrincipalRoleMap(dep)
     427        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
     428        self.assertEqual(user_name, 'anne')
     429        self.assertEqual(local_role, 'waeup.local.DepartmentManager')
     430        shutil.rmtree(os.path.dirname(fin_file))
     431        return
     432
     433class CourseProcessorTests(UniversityProcessorSetup):
    120434
    121435    def setUp(self):
    122436        super(CourseProcessorTests, self).setUp()
    123437        self.proc = CourseProcessor()
    124         self.site0 = dict()
    125         self.site1 = dict(faculties=dict())
    126         self.site2 = dict(faculties=dict(FAC=dict()))
    127         self.department1 = Department(code='DPT')
    128         self.department2 = Department(code='DPT')
    129         self.course = Course(code='CRS')
    130         self.department2.courses['CRS'] = self.course
    131         self.site3 = dict(faculties=dict(FAC=dict(DPT=self.department1)))
    132         self.site4 = dict(faculties=dict(FAC=dict(DPT=self.department2)))
    133         self.row = dict(department_code='DPT', faculty_code='FAC', code="CRS")
     438        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRS1")
     439        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRS1")
     440        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS2")
     441        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRS1")
     442
     443        self.csv_file_course = os.path.join(self.workdir, 'sample_course_data.csv')
     444        open(self.csv_file_course, 'wb').write(COURSE_SAMPLE_DATA)
    134445        return
    135446
    … …  
    142453    def test_parents_exist(self):
    143454        # make sure we lookup parents correctly
    144         result0 = self.proc.parentsExist(self.row, self.site0)
    145         result1 = self.proc.parentsExist(self.row, self.site1)
    146         result2 = self.proc.parentsExist(self.row, self.site2)
    147         result3 = self.proc.parentsExist(self.row, self.site3)
    148         result4 = self.proc.parentsExist(self.row, self.site4)
    149         self.assertTrue(result0 is False)
     455        result1 = self.proc.parentsExist(self.row1, self.app)
     456        result2 = self.proc.parentsExist(self.row2, self.app)
     457        result3 = self.proc.parentsExist(self.row3, self.app)
    150458        self.assertTrue(result1 is False)
    151459        self.assertTrue(result2 is False)
    152460        self.assertTrue(result3 is True)
    153         self.assertTrue(result4 is True)
    154461        return
    155462
    156463    def test_entry_exists(self):
    157464        # make sure we find an entry if it exists
    158         result0 = self.proc.entryExists(self.row, self.site0)
    159         result1 = self.proc.entryExists(self.row, self.site1)
    160         result2 = self.proc.entryExists(self.row, self.site2)
    161         result3 = self.proc.entryExists(self.row, self.site3)
    162         result4 = self.proc.entryExists(self.row, self.site4)
    163         self.assertTrue(result0 is False)
     465        result1 = self.proc.entryExists(self.row1, self.app)
     466        result2 = self.proc.entryExists(self.row2, self.app)
     467        result3 = self.proc.entryExists(self.row3, self.app)
     468        result4 = self.proc.entryExists(self.row4, self.app)
    164469        self.assertTrue(result1 is False)
    165470        self.assertTrue(result2 is False)
    … …  
    170475    def test_get_entry(self):
    171476        # make sure we can get an entry if it exists
    172         result1 = self.proc.getEntry(self.row, self.site3)
    173         result2 = self.proc.getEntry(self.row, self.site4)
     477        result1 = self.proc.getEntry(self.row1, self.app)
     478        result2 = self.proc.getEntry(self.row4, self.app)
    174479        self.assertTrue(result1 is None)
    175480        self.assertTrue(result2 is self.course)
    … …  
    178483    def test_del_entry(self):
    179484        # make sure we can delete entries
    180         self.assertTrue('CRS' in self.department2.courses.keys())
    181         self.proc.delEntry(self.row, self.site4)
    182         self.assertTrue('CRS' not in self.department2.courses.keys())
    183         return
    184 
    185 class CertificateProcessorTests(FunctionalTestCase):
    186 
    187     layer = FunctionalLayer
     485        self.assertTrue('CRS1' in self.app['faculties']['FAC1']['DEP1'].courses.keys())
     486        self.proc.delEntry(self.row4, self.app)
     487        self.assertTrue('CRS1' not in self.app['faculties']['FAC1']['DEP1'].courses.keys())
     488        return
     489
     490    def test_import(self):
     491        num, num_warns, fin_file, fail_file = self.proc.doImport(
     492            self.csv_file_course, COURSE_HEADER_FIELDS)
     493        content = open(fail_file).read()
     494        self.assertEqual(num_warns,6)
     495        self.assertEqual(
     496            content,
     497            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
     498            'FAC1,DEP1,CRS2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.Lecturer\'}]",'
     499            'local_roles: alice does not exist\r\n'
     500            'FAC1,DEP1,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
     501            'local_roles: waeup.local.Boss not allowed\r\n'
     502            'FAC1,DEP1,CRS2,[(\'anything\')],local_roles: no dicts\r\n'
     503            'FAC1,DEP1,CRS2,[,local_roles: Error\r\n'
     504            'FAC1,DEP1,CRS2,"[{\'user\':\'anne\',\'local\':\'waeup.local.Lecturer\'}]",'
     505            'local_roles: user_name or local_role missing\r\n'
     506            'FAC11,DEP2,CRS2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Lecturer\'}]",'
     507            'Not all parents do exist yet. Skipping\r\n'
     508            )
     509        # Anne got a local role in course CRS2.
     510        dep = self.app['faculties']['FAC1']['DEP1'].courses['CRS2']
     511        role_map = IPrincipalRoleMap(dep)
     512        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
     513        self.assertEqual(user_name, 'anne')
     514        self.assertEqual(local_role, 'waeup.local.Lecturer')
     515        shutil.rmtree(os.path.dirname(fin_file))
     516        return
     517
     518
     519class CertificateProcessorTests(UniversityProcessorSetup):
    188520
    189521    def setUp(self):
    190522        super(CertificateProcessorTests, self).setUp()
    191523        self.proc = CertificateProcessor()
    192         self.site0 = dict()
    193         self.site1 = dict(faculties=dict())
    194         self.site2 = dict(faculties=dict(FAC=dict()))
    195         self.department1 = Department(code='DPT')
    196         self.department2 = Department(code='DPT')
    197         self.certificate = Certificate(code='CRT')
    198         self.department2.certificates['CRT'] = self.certificate
    199         self.site3 = dict(faculties=dict(FAC=dict(DPT=self.department1)))
    200         self.site4 = dict(faculties=dict(FAC=dict(DPT=self.department2)))
    201         self.row = dict(department_code='DPT', faculty_code='FAC', code="CRT")
     524        self.row1 = dict(department_code='DEP2', faculty_code='FAC1', code="CRT1")
     525        self.row2 = dict(department_code='DEP1', faculty_code='FAC2', code="CRT1")
     526        self.row3 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT2")
     527        self.row4 = dict(department_code='DEP1', faculty_code='FAC1', code="CRT1")
     528
     529        self.csv_file_certificate = os.path.join(self.workdir, 'sample_certificate_data.csv')
     530        open(self.csv_file_certificate, 'wb').write(CERTIFICATE_SAMPLE_DATA)
    202531        return
    203532
    204533    def test_ifaces(self):
    205534        # Make sure we fullfill all interface contracts
    206         verifyClass(IBatchProcessor, CertificateProcessor)
     535        verifyClass(IBatchProcessor, CourseProcessor)
    207536        verifyObject(IBatchProcessor, self.proc)
    208537        return
    209538
    210539    def test_parents_exist(self):
    211         # make sure we can find all certificate parents
    212         result0 = self.proc.parentsExist(self.row, self.site0)
    213         result1 = self.proc.parentsExist(self.row, self.site1)
    214         result2 = self.proc.parentsExist(self.row, self.site2)
    215         result3 = self.proc.parentsExist(self.row, self.site3)
    216         result4 = self.proc.parentsExist(self.row, self.site4)
    217         self.assertTrue(result0 is False)
     540        # make sure we lookup parents correctly
     541        result1 = self.proc.parentsExist(self.row1, self.app)
     542        result2 = self.proc.parentsExist(self.row2, self.app)
     543        result3 = self.proc.parentsExist(self.row3, self.app)
    218544        self.assertTrue(result1 is False)
    219545        self.assertTrue(result2 is False)
    220546        self.assertTrue(result3 is True)
    221         self.assertTrue(result4 is True)
    222547        return
    223548
    224549    def test_entry_exists(self):
    225550        # make sure we find an entry if it exists
    226         result0 = self.proc.entryExists(self.row, self.site0)
    227         result1 = self.proc.entryExists(self.row, self.site1)
    228         result2 = self.proc.entryExists(self.row, self.site2)
    229         result3 = self.proc.entryExists(self.row, self.site3)
    230         result4 = self.proc.entryExists(self.row, self.site4)
    231         self.assertTrue(result0 is False)
     551        result1 = self.proc.entryExists(self.row1, self.app)
     552        result2 = self.proc.entryExists(self.row2, self.app)
     553        result3 = self.proc.entryExists(self.row3, self.app)
     554        result4 = self.proc.entryExists(self.row4, self.app)
    232555        self.assertTrue(result1 is False)
    233556        self.assertTrue(result2 is False)
    … …  
    238561    def test_get_entry(self):
    239562        # make sure we can get an entry if it exists
    240         result1 = self.proc.getEntry(self.row, self.site3)
    241         result2 = self.proc.getEntry(self.row, self.site4)
     563        result1 = self.proc.getEntry(self.row1, self.app)
     564        result2 = self.proc.getEntry(self.row4, self.app)
    242565        self.assertTrue(result1 is None)
    243566        self.assertTrue(result2 is self.certificate)
    … …  
    246569    def test_del_entry(self):
    247570        # make sure we can delete entries
    248         self.assertTrue('CRT' in self.department2.certificates.keys())
    249         self.proc.delEntry(self.row, self.site4)
    250         self.assertTrue('CRT' not in self.department2.certificates.keys())
    251         return
     571        self.assertTrue('CRT1' in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
     572        self.proc.delEntry(self.row4, self.app)
     573        self.assertTrue('CRT1' not in self.app['faculties']['FAC1']['DEP1'].certificates.keys())
     574        return
     575
     576    def test_import(self):
     577        num, num_warns, fin_file, fail_file = self.proc.doImport(
     578            self.csv_file_certificate, CERTIFICATE_HEADER_FIELDS)
     579        content = open(fail_file).read()
     580        self.assertEqual(num_warns,6)
     581        self.assertEqual(
     582            content,
     583            'faculty_code,department_code,code,local_roles,--ERRORS--\r\n'
     584            'FAC1,DEP1,CRT2,"[{\'user_name\':\'alice\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
     585            'local_roles: alice does not exist\r\n'
     586            'FAC1,DEP1,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.Boss\'}]",'
     587            'local_roles: waeup.local.Boss not allowed\r\n'
     588            'FAC1,DEP1,CRT2,[(\'anything\')],local_roles: no dicts\r\n'
     589            'FAC1,DEP1,CRT2,[,local_roles: Error\r\n'
     590            'FAC1,DEP1,CRT2,"[{\'user\':\'anne\',\'local\':\'waeup.local.CourseAdviser100\'}]",'
     591            'local_roles: user_name or local_role missing\r\n'
     592            'FAC11,DEP2,CRT2,"[{\'user_name\':\'anne\',\'local_role\':\'waeup.local.CourseAdviser100\'}]",'
     593            'Not all parents do exist yet. Skipping\r\n'
     594            )
     595        # Anne got a local role in certificate CRT2.
     596        dep = self.app['faculties']['FAC1']['DEP1'].certificates['CRT2']
     597        role_map = IPrincipalRoleMap(dep)
     598        local_role, user_name, setting = role_map.getPrincipalsAndRoles()[0]
     599        self.assertEqual(user_name, 'anne')
     600        self.assertEqual(local_role, 'waeup.local.CourseAdviser100')
     601        shutil.rmtree(os.path.dirname(fin_file))
     602       
     603        logcontent = open(self.logfile).read()
     604        # Logging message from updateEntry
     605        self.assertTrue(
     606            'INFO - system - Certificate Processor - CRT2 - Record updated: '
     607            'local_roles=anne|waeup.local.CourseAdviser100, code=CRT2'
     608            in logcontent)       
     609       
     610        return
     611
    252612
    253613class CertCourseProcessorTests(FunctionalTestCase):
    … …  
    266626        self.department2.certificates['CRT'] = self.certificate
    267627        self.course = Course(code='CRS')
    268         self.department2.certificates['CRT'].addCourseRef(self.course)
     628        self.department2.certificates['CRT'].addCertCourse(self.course)
    269629        self.cert_course = self.certificate['CRS_100']
    270630        self.department2.courses['CRS'] = self.course

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/test_export.py

    r8303 r9169  
    2222from zope.component import queryUtility
    2323from zope.interface.verify import verifyObject, verifyClass
     24from zope.securitypolicy.interfaces import IPrincipalRoleManager
    2425from waeup.kofa.interfaces import ICSVExporter
    25 from waeup.kofa.testing import KofaUnitTestLayer
     26from waeup.kofa.testing import KofaUnitTestLayer, FunctionalLayer
    2627from waeup.kofa.university import (
    2728    FacultiesContainer, Faculty, Department, Course, Certificate,
    … …  
    3435class FacultyExporterTest(unittest.TestCase):
    3536
    36     layer = KofaUnitTestLayer
     37    layer = FunctionalLayer
    3738
    3839    def setUp(self):
    … …  
    6667        self.assertEqual(
    6768            result,
    68             'code,title,title_prefix\r\n'
    69             'F1,Faculty of Cheese,faculty\r\n'
     69            'code,title,title_prefix,users_with_local_roles\r\n'
     70            'F1,Faculty of Cheese,faculty,[]\r\n'
    7071            )
    7172        return
    … …  
    7879        self.assertEqual(
    7980            result,
    80             'code,title,title_prefix\r\n'
    81             'F1,Faculty of Cheese,faculty\r\n'
     81            'code,title,title_prefix,users_with_local_roles\r\n'
     82            'F1,Faculty of Cheese,faculty,[]\r\n'
    8283            )
    8384        return
    … …  
    9697        self.assertEqual(
    9798            result,
    98             'code,title,title_prefix\r\n'
    99             'F1,Faculty of Cheese,faculty\r\n'
    100             'F2,Centre of Onion,centre\r\n'
     99            'code,title,title_prefix,users_with_local_roles\r\n'
     100            'F1,Faculty of Cheese,faculty,[]\r\n'
     101            'F2,Centre of Onion,centre,[]\r\n'
    101102            )
    102103        return
    … …  
    114115        self.assertEqual(
    115116            result,
    116             'code,title,title_prefix\r\n'
    117             'F1,Faculty of Cheese,faculty\r\n'
    118             'F2,Centre of Onion,centre\r\n'
     117            'code,title,title_prefix,users_with_local_roles\r\n'
     118            'F1,Faculty of Cheese,faculty,[]\r\n'
     119            'F2,Centre of Onion,centre,[]\r\n'
    119120            )
    120121        return
    … …  
    123124    # Tests for DepartmentExporter
    124125
    125     layer = KofaUnitTestLayer
     126    layer = FunctionalLayer
    126127
    127128    def setUp(self):
    … …  
    141142        self.fac1.addDepartment(self.dept2)
    142143        self.fac2.addDepartment(self.dept3)
     144        role_manager = IPrincipalRoleManager(self.dept1)
     145        role_manager.assignRoleToPrincipal(u'bobsrole', u'bob')
    143146        return
    144147
    … …  
    167170        self.assertEqual(
    168171            result,
    169             'code,faculty_code,title,title_prefix\r\n'
    170             'D1,F1,Department of Cheddar,department\r\n'
     172            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
     173            'D1,F1,Department of Cheddar,department,'
     174            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
    171175            )
    172176        return
    … …  
    178182        self.assertEqual(
    179183            result,
    180             'code,faculty_code,title,title_prefix\r\n'
    181             'D1,F1,Department of Cheddar,department\r\n'
    182             'D2,F1,Institue of Gouda,institute\r\n'
     184            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
     185            'D1,F1,Department of Cheddar,department,'
     186            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
     187            'D2,F1,Institue of Gouda,institute,[]\r\n'
    183188            )
    184189        return
    … …  
    191196        self.assertEqual(
    192197            result,
    193             'code,faculty_code,title,title_prefix\r\n'
    194             'D1,F1,Department of Cheddar,department\r\n'
    195             'D2,F1,Institue of Gouda,institute\r\n'
    196             'D3,F2,Department of Rings,department\r\n'
     198            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
     199            'D1,F1,Department of Cheddar,department,'
     200            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
     201            'D2,F1,Institue of Gouda,institute,[]\r\n'
     202            'D3,F2,Department of Rings,department,[]\r\n'
    197203            )
    198204        return
    … …  
    204210        self.assertEqual(
    205211            result,
    206             'code,faculty_code,title,title_prefix\r\n'
    207             'D1,F1,Department of Cheddar,department\r\n'
    208             'D2,F1,Institue of Gouda,institute\r\n'
    209             'D3,F2,Department of Rings,department\r\n'
     212            'code,faculty_code,title,title_prefix,users_with_local_roles\r\n'
     213            'D1,F1,Department of Cheddar,department,'
     214            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
     215            'D2,F1,Institue of Gouda,institute,[]\r\n'
     216            'D3,F2,Department of Rings,department,[]\r\n'
    210217            )
    211218        return
    … …  
    214221    # Tests for CourseExporter
    215222
    216     layer = KofaUnitTestLayer
     223    layer = FunctionalLayer
    217224
    218225    def setUp(self):
    … …  
    260267        self.assertEqual(
    261268            result,
    262             'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
    263             'C1,F1,D1,Cheese Basics,0,40,1\r\n'
     269            'code,faculty_code,department_code,title,credits,'
     270            'passmark,semester,users_with_local_roles\r\n'
     271            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
    264272            )
    265273        return
    … …  
    271279        self.assertEqual(
    272280            result,
    273             'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
    274             'C1,F1,D1,Cheese Basics,0,40,1\r\n'
    275             'C2,F1,D1,Advanced Cheese Making,0,40,1\r\n'
     281            'code,faculty_code,department_code,title,credits,passmark,'
     282            'semester,users_with_local_roles\r\n'
     283            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
     284            'C2,F1,D1,Advanced Cheese Making,0,40,1,[]\r\n'
    276285            )
    277286        return
    … …  
    284293        self.assertEqual(
    285294            result,
    286             'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
    287             'C1,F1,D1,Cheese Basics,0,40,1\r\n'
    288             'C2,F1,D1,Advanced Cheese Making,0,40,1\r\n'
    289             'C3,F1,D2,Selling Cheese,0,40,1\r\n'
     295            'code,faculty_code,department_code,title,credits,passmark,'
     296            'semester,users_with_local_roles\r\n'
     297            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
     298            'C2,F1,D1,Advanced Cheese Making,0,40,1,[]\r\n'
     299            'C3,F1,D2,Selling Cheese,0,40,1,[]\r\n'
    290300            )
    291301        return
    … …  
    297307        self.assertEqual(
    298308            result,
    299             'code,faculty_code,department_code,title,credits,passmark,semester\r\n'
    300             'C1,F1,D1,Cheese Basics,0,40,1\r\n'
    301             'C2,F1,D1,Advanced Cheese Making,0,40,1\r\n'
    302             'C3,F1,D2,Selling Cheese,0,40,1\r\n'
     309            'code,faculty_code,department_code,title,credits,passmark,'
     310            'semester,users_with_local_roles\r\n'
     311            'C1,F1,D1,Cheese Basics,0,40,1,[]\r\n'
     312            'C2,F1,D1,Advanced Cheese Making,0,40,1,[]\r\n'
     313            'C3,F1,D2,Selling Cheese,0,40,1,[]\r\n'
    303314            )
    304315        return
    … …  
    307318    # Tests for CertificateExporter
    308319
    309     layer = KofaUnitTestLayer
     320    layer = FunctionalLayer
    310321
    311322    def setUp(self):
    … …  
    339350        self.dept1.certificates.addCertificate(self.cert2)
    340351        self.dept2.certificates.addCertificate(self.cert3)
     352        role_manager = IPrincipalRoleManager(self.cert1)
     353        role_manager.assignRoleToPrincipal(u'bobsrole', u'bob')
    341354        return
    342355
    … …  
    365378        self.assertEqual(
    366379            result,
    367             'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
    368             'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
     380            'code,faculty_code,department_code,title,study_mode,start_level,'
     381            'end_level,application_category,school_fee_1,'
     382            'school_fee_2,school_fee_3,school_fee_4,users_with_local_roles\r\n'
     383            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
     384            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
    369385            )
    370386        return
    … …  
    376392        self.assertEqual(
    377393            result,
    378             'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
    379             'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
    380             'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,\r\n'
     394            'code,faculty_code,department_code,title,study_mode,start_level,'
     395            'end_level,application_category,school_fee_1,'
     396            'school_fee_2,school_fee_3,school_fee_4,users_with_local_roles\r\n'
     397            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
     398            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
     399            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,,,,[]\r\n'
    381400            )
    382401        return
    … …  
    389408        self.assertEqual(
    390409            result,
    391             'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
    392             'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
    393             'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,\r\n'
    394             'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,\r\n'
     410            'code,faculty_code,department_code,title,study_mode,start_level,'
     411            'end_level,application_category,'
     412            'school_fee_1,school_fee_2,school_fee_3,school_fee_4,'
     413            'users_with_local_roles\r\n'
     414            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
     415            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
     416            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,,,,[]\r\n'
     417            'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,,,,[]\r\n'
    395418            )
    396419        return
    … …  
    402425        self.assertEqual(
    403426            result,
    404             'code,faculty_code,department_code,title,study_mode,start_level,end_level,application_category,school_fee_1,school_fee_2\r\n'
    405             'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,\r\n'
    406             'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,\r\n'
    407             'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,\r\n'
     427            'code,faculty_code,department_code,title,study_mode,start_level,'
     428            'end_level,application_category,'
     429            'school_fee_1,school_fee_2,school_fee_3,school_fee_4,'
     430            'users_with_local_roles\r\n'
     431            'CERT1,F1,D1,Master of Cheese,ct_ft,100,300,basic,,,,,'
     432            '"[{\'user_name\': u\'bob\', \'local_role\': u\'bobsrole\'}]"\r\n'
     433            'CERT2,F1,D1,Master of Cheddar,ct_ft,400,700,cest,,,,,[]\r\n'
     434            'CERT3,F1,D2,Cert. of Rubbish,dp_pt,100,200,no,,,,,[]\r\n'
    408435            )
    409436        return
    … …  
    444471        self.dept1.certificates.addCertificate(self.cert2)
    445472        self.dept2.certificates.addCertificate(self.cert3)
    446         self.cert1.addCourseRef(self.course1, 100, True)
    447         self.cert1.addCourseRef(self.course2, 400, False)
    448         self.cert3.addCourseRef(self.course3, 100, False)
     473        self.cert1.addCertCourse(self.course1, 100, True)
     474        self.cert1.addCertCourse(self.course2, 400, False)
     475        self.cert3.addCertCourse(self.course3, 100, False)
    449476        self.certcourse1 = self.cert1['C1_100']
    450477        self.certcourse2 = self.cert1['C2_400']

  • main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/university/tests/test_university.py

    r7811 r9169  
    3333def test_suite():
    3434    # collect doctests for university subpackage
    35     suite = get_doctest_suite(['university/courserefs.txt',])
     35    suite = get_doctest_suite(['university/certcourses.txt',])
    3636    # add local unittests (actually only one)
    3737    for testcase in [UniversitySubpackageTests,]:
Note: See TracChangeset for help on using the changeset viewer.