Changeset 9294 for main/waeup.kofa/trunk/src/waeup/kofa/students
- Timestamp:
- 4 Oct 2012, 15:40:40 (12 years ago)
- File:
-
- 1 edited
Legend:
- Unmodified
- Added
- Removed
-
main/waeup.kofa/trunk/src/waeup/kofa/students/tests/test_batching.py
r9293 r9294 574 574 shutil.rmtree(os.path.dirname(fin_file)) 575 575 576 class StudentStudyLevelProcessorTest(StudentImportExportSetup): 577 578 def setUp(self): 579 super(StudentStudyLevelProcessorTest, self).setUp() 580 581 # Import students with subobjects 582 student_file = os.path.join(self.workdir, 'sample_student_data.csv') 583 open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) 584 num, num_warns, fin_file, fail_file = StudentProcessor().doImport( 585 student_file, STUDENT_HEADER_FIELDS) 586 shutil.rmtree(os.path.dirname(fin_file)) 587 588 # Update study courses 589 studycourse_file = os.path.join( 590 self.workdir, 'sample_studycourse_data.csv') 591 open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) 592 processor = StudentStudyCourseProcessor() 593 num, num_warns, fin_file, fail_file = processor.doImport( 594 studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') 595 shutil.rmtree(os.path.dirname(fin_file)) 596 597 self.processor = StudentStudyLevelProcessor() 598 self.csv_file = os.path.join( 599 self.workdir, 'sample_studylevel_data.csv') 600 open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) 601 602 def test_interface(self): 603 # Make sure we fulfill the interface contracts. 604 assert verifyObject(IBatchProcessor, self.processor) is True 605 assert verifyClass( 606 IBatchProcessor, StudentStudyLevelProcessor) is True 607 608 def test_checkConversion(self): 609 errs, inv_errs, conv_dict = self.processor.checkConversion( 610 dict(reg_number='1', level='220')) 611 self.assertEqual(len(errs),0) 612 errs, inv_errs, conv_dict = self.processor.checkConversion( 613 dict(reg_number='1', level='900')) 614 self.assertEqual(len(errs),1) 615 self.assertTrue(('level','no valid integer') in errs) 616 errs, inv_errs, conv_dict = self.processor.checkConversion( 617 dict(reg_number='1', level='xyz')) 618 self.assertEqual(len(errs),1) 619 self.assertTrue(('level','no integer') in errs) 620 621 def test_import(self): 622 num, num_warns, fin_file, fail_file = self.processor.doImport( 623 self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') 624 self.assertEqual(num_warns,2) 625 assert self.processor.entryExists( 626 dict(reg_number='1', level='100'), self.app) is True 627 studylevel = self.processor.getEntry( 628 dict(reg_number='1', level='100'), self.app) 629 self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1') 630 self.assertEqual(studylevel.level_session, 2008) 631 self.assertEqual(studylevel.level_verdict, None) 632 self.assertEqual(studylevel.level, 100) 633 shutil.rmtree(os.path.dirname(fin_file)) 634 635 logcontent = open(self.logfile).read() 636 # Logging message from updateEntry, 637 self.assertTrue( 638 'INFO - system - K1000000 - Study level ' 639 'updated: level=100, level_verdict=C, level_session=2009' 640 in logcontent) 641 642 def test_import_update(self): 643 # We perform the same import twice, 644 # the second time in update mode. The number 645 # of warnings must be the same. 646 num, num_warns, fin_file, fail_file = self.processor.doImport( 647 self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') 648 shutil.rmtree(os.path.dirname(fin_file)) 649 num, num_warns, fin_file, fail_file = self.processor.doImport( 650 self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update') 651 self.assertEqual(num_warns,2) 652 shutil.rmtree(os.path.dirname(fin_file)) 653 654 class CourseTicketProcessorTest(StudentImportExportSetup): 655 656 def setUp(self): 657 super(CourseTicketProcessorTest, self).setUp() 658 659 # Import students with subobjects 660 student_file = os.path.join(self.workdir, 'sample_student_data.csv') 661 open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) 662 num, num_warns, fin_file, fail_file = StudentProcessor().doImport( 663 student_file, STUDENT_HEADER_FIELDS) 664 shutil.rmtree(os.path.dirname(fin_file)) 665 666 # Add course and certificate course 667 self.course = createObject('waeup.Course') 668 self.course.code = 'COURSE1' 669 self.course.semester = 1 670 self.course.credits = 10 671 self.course.passmark = 40 672 self.app['faculties']['fac1']['dep1'].courses.addCourse( 673 self.course) 674 self.app['faculties']['fac1']['dep1'].certificates[ 675 'CERT1'].addCertCourse( 676 self.course, level=100) 677 678 # Update study courses 679 studycourse_file = os.path.join( 680 self.workdir, 'sample_studycourse_data.csv') 681 open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) 682 processor = StudentStudyCourseProcessor() 683 num, num_warns, fin_file, fail_file = processor.doImport( 684 studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') 685 shutil.rmtree(os.path.dirname(fin_file)) 686 687 # Import study levels 688 processor = StudentStudyLevelProcessor() 689 studylevel_file = os.path.join( 690 self.workdir, 'sample_studylevel_data.csv') 691 open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) 692 num, num_warns, fin_file, fail_file = processor.doImport( 693 studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create') 694 shutil.rmtree(os.path.dirname(fin_file)) 695 696 self.processor = CourseTicketProcessor() 697 self.csv_file = os.path.join( 698 self.workdir, 'sample_courseticket_data.csv') 699 open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA) 700 701 def test_interface(self): 702 # Make sure we fulfill the interface contracts. 703 assert verifyObject(IBatchProcessor, self.processor) is True 704 assert verifyClass( 705 IBatchProcessor, CourseTicketProcessor) is True 706 707 def test_checkConversion(self): 708 errs, inv_errs, conv_dict = self.processor.checkConversion( 709 dict(reg_number='1', code='COURSE1', level='220')) 710 self.assertEqual(len(errs),0) 711 errs, inv_errs, conv_dict = self.processor.checkConversion( 712 dict(reg_number='1', code='COURSE2', level='220')) 713 self.assertEqual(len(errs),1) 714 self.assertTrue(('code','non-existent') in errs) 715 716 def test_import(self): 717 718 num, num_warns, fin_file, fail_file = self.processor.doImport( 719 self.csv_file, COURSETICKET_HEADER_FIELDS,'create') 720 721 self.assertEqual(num_warns,2) 722 assert self.processor.entryExists( 723 dict(reg_number='1', level='100', code='COURSE1'), 724 self.app) is True 725 courseticket = self.processor.getEntry( 726 dict(reg_number='1', level='100', code='COURSE1'), self.app) 727 self.assertEqual(courseticket.__parent__.__parent__.certificate.code, 728 u'CERT1') 729 self.assertEqual(courseticket.score, 1) 730 self.assertEqual(courseticket.mandatory, True) 731 self.assertEqual(courseticket.fcode, 'NA') 732 self.assertEqual(courseticket.dcode, 'NA') 733 self.assertEqual(courseticket.code, 'COURSE1') 734 self.assertEqual(courseticket.title, 'Unnamed Course') 735 self.assertEqual(courseticket.credits, 10) 736 self.assertEqual(courseticket.passmark, 40) 737 self.assertEqual(courseticket.semester, 1) 738 shutil.rmtree(os.path.dirname(fin_file)) 739 740 logcontent = open(self.logfile).read() 741 # Logging message from updateEntry, 742 self.assertTrue( 743 'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, ' 744 'mandatory=False, score=3' 745 in logcontent) 746 747 def test_import_update(self): 748 # We perform the same import twice, 749 # the second time in update mode. The number 750 # of warnings must be the same. 751 num, num_warns, fin_file, fail_file = self.processor.doImport( 752 self.csv_file, COURSETICKET_HEADER_FIELDS,'create') 753 shutil.rmtree(os.path.dirname(fin_file)) 754 num, num_warns, fin_file, fail_file = self.processor.doImport( 755 self.csv_file, COURSETICKET_HEADER_FIELDS,'update') 756 self.assertEqual(num_warns,2) 757 shutil.rmtree(os.path.dirname(fin_file)) 758 759 def test_import_remove(self): 760 # We perform the same import twice, 761 # the second time in remove mode. The number 762 # of warnings must be the same. 763 num, num_warns, fin_file, fail_file = self.processor.doImport( 764 self.csv_file, COURSETICKET_HEADER_FIELDS,'create') 765 shutil.rmtree(os.path.dirname(fin_file)) 766 assert self.processor.entryExists( 767 dict(reg_number='1', level='100', code='COURSE1'), self.app) is True 768 num, num_warns, fin_file, fail_file = self.processor.doImport( 769 self.csv_file, COURSETICKET_HEADER_FIELDS,'remove') 770 self.assertEqual(num_warns,2) 771 assert self.processor.entryExists( 772 dict(reg_number='1', level='100', code='COURSE1'), self.app) is False 773 shutil.rmtree(os.path.dirname(fin_file)) 774 logcontent = open(self.logfile).read() 775 self.assertTrue( 776 'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1' 777 in logcontent) 778 779 class PaymentProcessorTest(StudentImportExportSetup): 780 781 def setUp(self): 782 super(PaymentProcessorTest, self).setUp() 783 784 # Add student with payment 785 student = Student() 786 student.firstname = u'Anna' 787 student.lastname = u'Tester' 788 student.reg_number = u'123' 789 student.matric_number = u'234' 790 self.app['students'].addStudent(student) 791 self.student = self.app['students'][student.student_id] 792 payment = createObject(u'waeup.StudentOnlinePayment') 793 payment.p_id = 'p120' 794 self.student['payments'][payment.p_id] = payment 795 796 # Import students with subobjects 797 student_file = os.path.join(self.workdir, 'sample_student_data.csv') 798 open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) 799 num, num_warns, fin_file, fail_file = StudentProcessor().doImport( 800 student_file, STUDENT_HEADER_FIELDS) 801 shutil.rmtree(os.path.dirname(fin_file)) 802 803 self.processor = StudentOnlinePaymentProcessor() 804 self.csv_file = os.path.join( 805 self.workdir, 'sample_payment_data.csv') 806 open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA) 807 self.csv_file2 = os.path.join( 808 self.workdir, 'sample_create_payment_data.csv') 809 open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA) 810 811 def test_interface(self): 812 # Make sure we fulfill the interface contracts. 813 assert verifyObject(IBatchProcessor, self.processor) is True 814 assert verifyClass( 815 IBatchProcessor, StudentOnlinePaymentProcessor) is True 816 817 def test_getEntry(self): 818 assert self.processor.getEntry( 819 dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None 820 assert self.processor.getEntry( 821 dict(student_id=self.student.student_id, p_id='p120'), 822 self.app) is self.student['payments']['p120'] 823 assert self.processor.getEntry( 824 dict(student_id=self.student.student_id, p_id='XXXXXX112'), 825 self.app) is self.student['payments']['p120'] 826 827 def test_delEntry(self): 828 assert self.processor.getEntry( 829 dict(student_id=self.student.student_id, p_id='p120'), 830 self.app) is self.student['payments']['p120'] 831 self.assertEqual(len(self.student['payments'].keys()),1) 832 self.processor.delEntry( 833 dict(student_id=self.student.student_id, p_id='p120'), 834 self.app) 835 assert self.processor.getEntry( 836 dict(student_id=self.student.student_id, p_id='p120'), 837 self.app) is None 838 self.assertEqual(len(self.student['payments'].keys()),0) 839 840 def test_addEntry(self): 841 self.assertEqual(len(self.student['payments'].keys()),1) 842 payment1 = createObject(u'waeup.StudentOnlinePayment') 843 payment1.p_id = 'p234' 844 self.processor.addEntry( 845 payment1, dict(student_id=self.student.student_id, p_id='p234'), 846 self.app) 847 self.assertEqual(len(self.student['payments'].keys()),2) 848 self.assertEqual(self.student['payments']['p234'].p_id, 'p234') 849 payment2 = createObject(u'waeup.StudentOnlinePayment') 850 payment1.p_id = 'nonsense' 851 # payment1.p_id will be replaced if p_id doesn't start with 'p' 852 self.processor.addEntry( 853 payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'), 854 self.app) 855 self.assertEqual(len(self.student['payments'].keys()),3) 856 self.assertEqual(self.student['payments']['p560'].p_id, 'p560') 857 858 def test_checkConversion(self): 859 errs, inv_errs, conv_dict = self.processor.checkConversion( 860 dict(p_id='3816951266236341955')) 861 self.assertEqual(len(errs),0) 862 errs, inv_errs, conv_dict = self.processor.checkConversion( 863 dict(p_id='p1266236341955')) 864 self.assertEqual(len(errs),0) 865 errs, inv_errs, conv_dict = self.processor.checkConversion( 866 dict(p_id='nonsense')) 867 self.assertEqual(len(errs),1) 868 timestamp = ("%d" % int(time()*10000))[1:] 869 p_id = "p%s" % timestamp 870 errs, inv_errs, conv_dict = self.processor.checkConversion( 871 dict(p_id=p_id)) 872 self.assertEqual(len(errs),0) 873 874 def test_import(self): 875 num, num_warns, fin_file, fail_file = self.processor.doImport( 876 self.csv_file, PAYMENT_HEADER_FIELDS,'create') 877 self.assertEqual(num_warns,0) 878 payment = self.processor.getEntry(dict(reg_number='1', 879 p_id='p2907979737440'), self.app) 880 self.assertEqual(payment.p_id, 'p2907979737440') 881 self.assertTrue(payment.p_current) 882 cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S") 883 self.assertEqual(cdate, "2010-11-26 18:59:33") 884 self.assertEqual(str(payment.creation_date.tzinfo),'UTC') 885 payment = self.processor.getEntry(dict(matric_number='100001', 886 p_id='p2907125937570'), self.app) 887 self.assertEqual(payment.p_id, 'p2907125937570') 888 self.assertFalse(payment.p_current) 889 cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S") 890 # Ooooh, still the old problem, see 891 # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html. 892 # WAT is interpreted as GMT-1 and not GMT+1 893 self.assertEqual(cdate, "2010-11-25 21:16:33") 894 self.assertEqual(str(payment.creation_date.tzinfo),'UTC') 895 shutil.rmtree(os.path.dirname(fin_file)) 896 logcontent = open(self.logfile).read() 897 # Logging message from updateEntry 898 self.assertTrue( 899 'INFO - system - K1000001 - Payment ticket updated: ' 900 'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, ' 901 'p_category=schoolfee, amount_auth=19500.0, p_current=True, ' 902 'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, ' 903 'p_state=paid' 904 in logcontent) 905 906 def test_import_update(self): 907 # We perform the same import twice, 908 # the second time in update mode. The number 909 # of warnings must be the same. 910 num, num_warns, fin_file, fail_file = self.processor.doImport( 911 self.csv_file, PAYMENT_HEADER_FIELDS,'create') 912 shutil.rmtree(os.path.dirname(fin_file)) 913 num, num_warns, fin_file, fail_file = self.processor.doImport( 914 self.csv_file, PAYMENT_HEADER_FIELDS,'update') 915 self.assertEqual(num_warns,0) 916 shutil.rmtree(os.path.dirname(fin_file)) 917 918 def test_import_remove(self): 919 num, num_warns, fin_file, fail_file = self.processor.doImport( 920 self.csv_file, PAYMENT_HEADER_FIELDS,'create') 921 shutil.rmtree(os.path.dirname(fin_file)) 922 num, num_warns, fin_file, fail_file = self.processor.doImport( 923 self.csv_file, PAYMENT_HEADER_FIELDS,'remove') 924 self.assertEqual(num_warns,0) 925 shutil.rmtree(os.path.dirname(fin_file)) 926 logcontent = open(self.logfile).read() 927 self.assertTrue( 928 'INFO - system - K1000001 - Payment ticket removed: p1266236341955' 929 in logcontent) 930 931 def test_import_wo_pid(self): 932 num, num_warns, fin_file, fail_file = self.processor.doImport( 933 self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create') 934 self.assertEqual(num_warns,0) 935 shutil.rmtree(os.path.dirname(fin_file)) 936 self.assertEqual(len(self.app['students']['X666666']['payments']), 50) 937 576 938 class StudentVerdictProcessorTest(StudentImportExportSetup): 577 939 … … 640 1002 '2008,A,True,100007,200,Study level object is missing.\r\n' 641 1003 ) 642 shutil.rmtree(os.path.dirname(fin_file))643 644 645 class StudentStudyLevelProcessorTest(StudentImportExportSetup):646 647 def setUp(self):648 super(StudentStudyLevelProcessorTest, self).setUp()649 650 # Import students with subobjects651 student_file = os.path.join(self.workdir, 'sample_student_data.csv')652 open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)653 num, num_warns, fin_file, fail_file = StudentProcessor().doImport(654 student_file, STUDENT_HEADER_FIELDS)655 shutil.rmtree(os.path.dirname(fin_file))656 657 # Update study courses658 studycourse_file = os.path.join(659 self.workdir, 'sample_studycourse_data.csv')660 open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)661 processor = StudentStudyCourseProcessor()662 num, num_warns, fin_file, fail_file = processor.doImport(663 studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')664 shutil.rmtree(os.path.dirname(fin_file))665 666 self.processor = StudentStudyLevelProcessor()667 self.csv_file = os.path.join(668 self.workdir, 'sample_studylevel_data.csv')669 open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)670 671 def test_interface(self):672 # Make sure we fulfill the interface contracts.673 assert verifyObject(IBatchProcessor, self.processor) is True674 assert verifyClass(675 IBatchProcessor, StudentStudyLevelProcessor) is True676 677 def test_checkConversion(self):678 errs, inv_errs, conv_dict = self.processor.checkConversion(679 dict(reg_number='1', level='220'))680 self.assertEqual(len(errs),0)681 errs, inv_errs, conv_dict = self.processor.checkConversion(682 dict(reg_number='1', level='900'))683 self.assertEqual(len(errs),1)684 self.assertTrue(('level','no valid integer') in errs)685 errs, inv_errs, conv_dict = self.processor.checkConversion(686 dict(reg_number='1', level='xyz'))687 self.assertEqual(len(errs),1)688 self.assertTrue(('level','no integer') in errs)689 690 def test_import(self):691 num, num_warns, fin_file, fail_file = self.processor.doImport(692 self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')693 self.assertEqual(num_warns,2)694 assert self.processor.entryExists(695 dict(reg_number='1', level='100'), self.app) is True696 studylevel = self.processor.getEntry(697 dict(reg_number='1', level='100'), self.app)698 self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')699 self.assertEqual(studylevel.level_session, 2008)700 self.assertEqual(studylevel.level_verdict, None)701 self.assertEqual(studylevel.level, 100)702 shutil.rmtree(os.path.dirname(fin_file))703 704 1004 logcontent = open(self.logfile).read() 705 # Logging message from updateEntry, 706 self.assertTrue( 707 'INFO - system - K1000000 - Study level ' 708 'updated: level=100, level_verdict=C, level_session=2009' 709 in logcontent) 710 711 def test_import_update(self): 712 # We perform the same import twice, 713 # the second time in update mode. The number 714 # of warnings must be the same. 715 num, num_warns, fin_file, fail_file = self.processor.doImport( 716 self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') 717 shutil.rmtree(os.path.dirname(fin_file)) 718 num, num_warns, fin_file, fail_file = self.processor.doImport( 719 self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update') 720 self.assertEqual(num_warns,2) 721 shutil.rmtree(os.path.dirname(fin_file)) 722 723 class CourseTicketProcessorTest(StudentImportExportSetup): 724 725 def setUp(self): 726 super(CourseTicketProcessorTest, self).setUp() 727 728 # Import students with subobjects 729 student_file = os.path.join(self.workdir, 'sample_student_data.csv') 730 open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) 731 num, num_warns, fin_file, fail_file = StudentProcessor().doImport( 732 student_file, STUDENT_HEADER_FIELDS) 733 shutil.rmtree(os.path.dirname(fin_file)) 734 735 # Add course and certificate course 736 self.course = createObject('waeup.Course') 737 self.course.code = 'COURSE1' 738 self.course.semester = 1 739 self.course.credits = 10 740 self.course.passmark = 40 741 self.app['faculties']['fac1']['dep1'].courses.addCourse( 742 self.course) 743 self.app['faculties']['fac1']['dep1'].certificates[ 744 'CERT1'].addCertCourse( 745 self.course, level=100) 746 747 # Update study courses 748 studycourse_file = os.path.join( 749 self.workdir, 'sample_studycourse_data.csv') 750 open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) 751 processor = StudentStudyCourseProcessor() 752 num, num_warns, fin_file, fail_file = processor.doImport( 753 studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') 754 shutil.rmtree(os.path.dirname(fin_file)) 755 756 # Import study levels 757 processor = StudentStudyLevelProcessor() 758 studylevel_file = os.path.join( 759 self.workdir, 'sample_studylevel_data.csv') 760 open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) 761 num, num_warns, fin_file, fail_file = processor.doImport( 762 studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create') 763 shutil.rmtree(os.path.dirname(fin_file)) 764 765 self.processor = CourseTicketProcessor() 766 self.csv_file = os.path.join( 767 self.workdir, 'sample_courseticket_data.csv') 768 open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA) 769 770 def test_interface(self): 771 # Make sure we fulfill the interface contracts. 772 assert verifyObject(IBatchProcessor, self.processor) is True 773 assert verifyClass( 774 IBatchProcessor, CourseTicketProcessor) is True 775 776 def test_checkConversion(self): 777 errs, inv_errs, conv_dict = self.processor.checkConversion( 778 dict(reg_number='1', code='COURSE1', level='220')) 779 self.assertEqual(len(errs),0) 780 errs, inv_errs, conv_dict = self.processor.checkConversion( 781 dict(reg_number='1', code='COURSE2', level='220')) 782 self.assertEqual(len(errs),1) 783 self.assertTrue(('code','non-existent') in errs) 784 785 def test_import(self): 786 787 num, num_warns, fin_file, fail_file = self.processor.doImport( 788 self.csv_file, COURSETICKET_HEADER_FIELDS,'create') 789 790 self.assertEqual(num_warns,2) 791 assert self.processor.entryExists( 792 dict(reg_number='1', level='100', code='COURSE1'), 793 self.app) is True 794 courseticket = self.processor.getEntry( 795 dict(reg_number='1', level='100', code='COURSE1'), self.app) 796 self.assertEqual(courseticket.__parent__.__parent__.certificate.code, 797 u'CERT1') 798 self.assertEqual(courseticket.score, 1) 799 self.assertEqual(courseticket.mandatory, True) 800 self.assertEqual(courseticket.fcode, 'NA') 801 self.assertEqual(courseticket.dcode, 'NA') 802 self.assertEqual(courseticket.code, 'COURSE1') 803 self.assertEqual(courseticket.title, 'Unnamed Course') 804 self.assertEqual(courseticket.credits, 10) 805 self.assertEqual(courseticket.passmark, 40) 806 self.assertEqual(courseticket.semester, 1) 807 shutil.rmtree(os.path.dirname(fin_file)) 808 809 logcontent = open(self.logfile).read() 810 # Logging message from updateEntry, 811 self.assertTrue( 812 'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, ' 813 'mandatory=False, score=3' 814 in logcontent) 815 816 def test_import_update(self): 817 # We perform the same import twice, 818 # the second time in update mode. The number 819 # of warnings must be the same. 820 num, num_warns, fin_file, fail_file = self.processor.doImport( 821 self.csv_file, COURSETICKET_HEADER_FIELDS,'create') 822 shutil.rmtree(os.path.dirname(fin_file)) 823 num, num_warns, fin_file, fail_file = self.processor.doImport( 824 self.csv_file, COURSETICKET_HEADER_FIELDS,'update') 825 self.assertEqual(num_warns,2) 826 shutil.rmtree(os.path.dirname(fin_file)) 827 828 def test_import_remove(self): 829 # We perform the same import twice, 830 # the second time in remove mode. The number 831 # of warnings must be the same. 832 num, num_warns, fin_file, fail_file = self.processor.doImport( 833 self.csv_file, COURSETICKET_HEADER_FIELDS,'create') 834 shutil.rmtree(os.path.dirname(fin_file)) 835 assert self.processor.entryExists( 836 dict(reg_number='1', level='100', code='COURSE1'), self.app) is True 837 num, num_warns, fin_file, fail_file = self.processor.doImport( 838 self.csv_file, COURSETICKET_HEADER_FIELDS,'remove') 839 self.assertEqual(num_warns,2) 840 assert self.processor.entryExists( 841 dict(reg_number='1', level='100', code='COURSE1'), self.app) is False 842 shutil.rmtree(os.path.dirname(fin_file)) 843 logcontent = open(self.logfile).read() 844 self.assertTrue( 845 'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1' 846 in logcontent) 847 848 class PaymentProcessorTest(StudentImportExportSetup): 849 850 def setUp(self): 851 super(PaymentProcessorTest, self).setUp() 852 853 # Add student with payment 854 student = Student() 855 student.firstname = u'Anna' 856 student.lastname = u'Tester' 857 student.reg_number = u'123' 858 student.matric_number = u'234' 859 self.app['students'].addStudent(student) 860 self.student = self.app['students'][student.student_id] 861 payment = createObject(u'waeup.StudentOnlinePayment') 862 payment.p_id = 'p120' 863 self.student['payments'][payment.p_id] = payment 864 865 # Import students with subobjects 866 student_file = os.path.join(self.workdir, 'sample_student_data.csv') 867 open(student_file, 'wb').write(STUDENT_SAMPLE_DATA) 868 num, num_warns, fin_file, fail_file = StudentProcessor().doImport( 869 student_file, STUDENT_HEADER_FIELDS) 870 shutil.rmtree(os.path.dirname(fin_file)) 871 872 self.processor = StudentOnlinePaymentProcessor() 873 self.csv_file = os.path.join( 874 self.workdir, 'sample_payment_data.csv') 875 open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA) 876 self.csv_file2 = os.path.join( 877 self.workdir, 'sample_create_payment_data.csv') 878 open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA) 879 880 def test_interface(self): 881 # Make sure we fulfill the interface contracts. 882 assert verifyObject(IBatchProcessor, self.processor) is True 883 assert verifyClass( 884 IBatchProcessor, StudentOnlinePaymentProcessor) is True 885 886 def test_getEntry(self): 887 assert self.processor.getEntry( 888 dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None 889 assert self.processor.getEntry( 890 dict(student_id=self.student.student_id, p_id='p120'), 891 self.app) is self.student['payments']['p120'] 892 assert self.processor.getEntry( 893 dict(student_id=self.student.student_id, p_id='XXXXXX112'), 894 self.app) is self.student['payments']['p120'] 895 896 def test_delEntry(self): 897 assert self.processor.getEntry( 898 dict(student_id=self.student.student_id, p_id='p120'), 899 self.app) is self.student['payments']['p120'] 900 self.assertEqual(len(self.student['payments'].keys()),1) 901 self.processor.delEntry( 902 dict(student_id=self.student.student_id, p_id='p120'), 903 self.app) 904 assert self.processor.getEntry( 905 dict(student_id=self.student.student_id, p_id='p120'), 906 self.app) is None 907 self.assertEqual(len(self.student['payments'].keys()),0) 908 909 def test_addEntry(self): 910 self.assertEqual(len(self.student['payments'].keys()),1) 911 payment1 = createObject(u'waeup.StudentOnlinePayment') 912 payment1.p_id = 'p234' 913 self.processor.addEntry( 914 payment1, dict(student_id=self.student.student_id, p_id='p234'), 915 self.app) 916 self.assertEqual(len(self.student['payments'].keys()),2) 917 self.assertEqual(self.student['payments']['p234'].p_id, 'p234') 918 payment2 = createObject(u'waeup.StudentOnlinePayment') 919 payment1.p_id = 'nonsense' 920 # payment1.p_id will be replaced if p_id doesn't start with 'p' 921 self.processor.addEntry( 922 payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'), 923 self.app) 924 self.assertEqual(len(self.student['payments'].keys()),3) 925 self.assertEqual(self.student['payments']['p560'].p_id, 'p560') 926 927 def test_checkConversion(self): 928 errs, inv_errs, conv_dict = self.processor.checkConversion( 929 dict(p_id='3816951266236341955')) 930 self.assertEqual(len(errs),0) 931 errs, inv_errs, conv_dict = self.processor.checkConversion( 932 dict(p_id='p1266236341955')) 933 self.assertEqual(len(errs),0) 934 errs, inv_errs, conv_dict = self.processor.checkConversion( 935 dict(p_id='nonsense')) 936 self.assertEqual(len(errs),1) 937 timestamp = ("%d" % int(time()*10000))[1:] 938 p_id = "p%s" % timestamp 939 errs, inv_errs, conv_dict = self.processor.checkConversion( 940 dict(p_id=p_id)) 941 self.assertEqual(len(errs),0) 942 943 def test_import(self): 944 num, num_warns, fin_file, fail_file = self.processor.doImport( 945 self.csv_file, PAYMENT_HEADER_FIELDS,'create') 946 self.assertEqual(num_warns,0) 947 payment = self.processor.getEntry(dict(reg_number='1', 948 p_id='p2907979737440'), self.app) 949 self.assertEqual(payment.p_id, 'p2907979737440') 950 self.assertTrue(payment.p_current) 951 cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S") 952 self.assertEqual(cdate, "2010-11-26 18:59:33") 953 self.assertEqual(str(payment.creation_date.tzinfo),'UTC') 954 payment = self.processor.getEntry(dict(matric_number='100001', 955 p_id='p2907125937570'), self.app) 956 self.assertEqual(payment.p_id, 'p2907125937570') 957 self.assertFalse(payment.p_current) 958 cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S") 959 # Ooooh, still the old problem, see 960 # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html. 961 # WAT is interpreted as GMT-1 and not GMT+1 962 self.assertEqual(cdate, "2010-11-25 21:16:33") 963 self.assertEqual(str(payment.creation_date.tzinfo),'UTC') 964 shutil.rmtree(os.path.dirname(fin_file)) 965 logcontent = open(self.logfile).read() 966 # Logging message from updateEntry 967 self.assertTrue( 968 'INFO - system - K1000001 - Payment ticket updated: ' 969 'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, ' 970 'p_category=schoolfee, amount_auth=19500.0, p_current=True, ' 971 'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, ' 972 'p_state=paid' 973 in logcontent) 974 975 def test_import_update(self): 976 # We perform the same import twice, 977 # the second time in update mode. The number 978 # of warnings must be the same. 979 num, num_warns, fin_file, fail_file = self.processor.doImport( 980 self.csv_file, PAYMENT_HEADER_FIELDS,'create') 981 shutil.rmtree(os.path.dirname(fin_file)) 982 num, num_warns, fin_file, fail_file = self.processor.doImport( 983 self.csv_file, PAYMENT_HEADER_FIELDS,'update') 984 self.assertEqual(num_warns,0) 985 shutil.rmtree(os.path.dirname(fin_file)) 986 987 def test_import_remove(self): 988 num, num_warns, fin_file, fail_file = self.processor.doImport( 989 self.csv_file, PAYMENT_HEADER_FIELDS,'create') 990 shutil.rmtree(os.path.dirname(fin_file)) 991 num, num_warns, fin_file, fail_file = self.processor.doImport( 992 self.csv_file, PAYMENT_HEADER_FIELDS,'remove') 993 self.assertEqual(num_warns,0) 994 shutil.rmtree(os.path.dirname(fin_file)) 995 logcontent = open(self.logfile).read() 996 self.assertTrue( 997 'INFO - system - K1000001 - Payment ticket removed: p1266236341955' 998 in logcontent) 999 1000 def test_import_wo_pid(self): 1001 num, num_warns, fin_file, fail_file = self.processor.doImport( 1002 self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create') 1003 self.assertEqual(num_warns,0) 1004 shutil.rmtree(os.path.dirname(fin_file)) 1005 self.assertEqual(len(self.app['students']['X666666']['payments']), 50) 1006 1005 self.assertMatches( 1006 '... INFO - system - X666666 - Study course updated: current_verdict=0...', 1007 logcontent) 1008 self.assertMatches( 1009 '... INFO - system - X666666 - Returned...', 1010 logcontent) 1011 1012 shutil.rmtree(os.path.dirname(fin_file)) 1007 1013 1008 1014 def test_suite():
Note: See TracChangeset for help on using the changeset viewer.