- Timestamp:
- 30 May 2011, 23:05:49 (14 years ago)
- File:
-
- 1 edited
Legend:
- Unmodified
- Added
- Removed
-
main/waeup.sirp/trunk/src/waeup/sirp/applicants/tests/test_batching.py
r6251 r6252 33 33 from waeup.sirp.app import University 34 34 from waeup.sirp.applicants.batching import ApplicantsContainerImporter 35 from waeup.sirp.applicants.container import ApplicantsContainer 35 36 from waeup.sirp.applicants.interfaces import IApplicantsContainer 36 37 from waeup.sirp.interfaces import DuplicationError … … 40 41 41 42 # Sample data we can use in tests... 42 JAMB_SAMPLE_DATA = '''reg_no,fst_sit_fname,lastname,firstname,middlenames,sex,date_of_birth,jamb_state,jamb_lga,course1,screening_date,screening_venue,entry_session,screening_type,ignore_this_col 43 91100546DD,ISUOSUO MOSES ODOMERO,ISUOSUO,MOSES,ODOMERO,M,25/5/1982,DEL,ISO-S,BSCPOL,2009/09/25 09:00:00 GMT+1,REPRINT SLIP AS FROM WED 23/09/2009,9,pde,blah 44 91111834CC,DURUIHEOMA AUGUSTINA ADANNA,DURUIHEOMA,AUGUSTINA,ADANNA,F,15/4/1986,IMO,MBAIT,BSCPOL,2009/09/25 09:00:00 GMT+1,REPRINT SLIP AS FROM WED 23/09/2009,9,pde,blah 45 91109351AC,ARISERE EBIKEBUNA COMFORT,ARISERE,EBIKEBUNA,COMFORT,F,6/1/1984,EDO,OV-SW,BSCPOL,2009/09/25 09:00:00 GMT+1,REPRINT SLIP AS FROM WED 23/09/2009,9,pde,blah 46 ''' 43 APPS_CONTAINER_SAMPLE_DATA = open( 44 os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'), 45 'rb').read() 47 46 48 47 # The header fields of the above CSV snippet 49 HEADER_FIELDS = JAMB_SAMPLE_DATA.split('\n')[0].split(',') 48 APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split( 49 '\n')[0].split(',') 50 50 51 class FakeSite(dict): 52 def getSiteManager(self): 53 return object() 51 class ApplicantsContainerImporterTest(FunctionalTestCase): 54 52 55 class ApplicantsContainerImporterTest(unittest.TestCase): 53 layer = FunctionalLayer 56 54 57 55 def setUp(self): 56 super(ApplicantsContainerImporterTest, self).setUp() 57 58 # Setup a sample site for each test 59 app = University() 60 self.dc_root = tempfile.mkdtemp() 61 app['datacenter'].setStoragePath(self.dc_root) 62 63 # Prepopulate the ZODB... 64 self.getRootFolder()['app'] = app 65 self.app = self.getRootFolder()['app'] 66 self.container = ApplicantsContainer() 67 self.container.code = u'dp2011' 68 self.app['applicants']['dp2011'] = self.container 69 58 70 self.importer = ApplicantsContainerImporter() 59 self.site1 = dict(60 jambdata = {61 'REG_NO_1': 'Application1',62 }63 )64 71 self.workdir = tempfile.mkdtemp() 65 72 self.csv_file = os.path.join(self.workdir, 'sampledata.csv') 66 open(self.csv_file, 'wb').write(JAMB_SAMPLE_DATA) 73 open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA) 74 setSite(self.app) 67 75 return 68 76 69 77 def tearDown(self): 78 super(ApplicantsContainerImporterTest, self).tearDown() 70 79 shutil.rmtree(self.workdir) 80 shutil.rmtree(self.dc_root) 71 81 clearSite() 72 82 return 73 83 74 84 def test_interface(self): 75 """Make sure we fulfill the interface contracts."""85 # Make sure we fulfill the interface contracts. 76 86 assert verifyObject(IBatchProcessor, self.importer) is True 77 87 assert verifyClass( … … 80 90 def test_parentsExist(self): 81 91 assert self.importer.parentsExist(None, dict()) is False 82 assert self.importer.parentsExist(None, self. site1) is True92 assert self.importer.parentsExist(None, self.app) is True 83 93 84 94 def test_entryExists(self): 85 95 assert self.importer.entryExists( 86 dict( reg_no='REG_NONE'), self.site1) is False96 dict(code='REG_NONE'), self.app) is False 87 97 assert self.importer.entryExists( 88 dict( reg_no='REG_NO_1'), self.site1) is True98 dict(code='dp2011'), self.app) is True 89 99 90 100 def test_getParent(self): 91 parent = self.importer.getParent(None, self. site1)92 assert parent is self. site1['jambdata']101 parent = self.importer.getParent(None, self.app) 102 assert parent is self.app['applicants'] 93 103 94 104 def test_getEntry(self): 95 105 assert self.importer.getEntry( 96 dict( reg_no='REG_NONE'), self.site1) is None106 dict(code='REG_NONE'), self.app) is None 97 107 assert self.importer.getEntry( 98 dict( reg_no='REG_NO_1'), self.site1) == 'Application1'108 dict(code='dp2011'), self.app) is self.container 99 109 100 110 def test_addEntry(self): 101 111 self.importer.addEntry( 102 'New application', dict( reg_no='REG_NO_99'), self.site1)103 assert self. site1['jambdata']['REG_NO_99'] == 'New application'112 'New application', dict(code='dp2012'), self.app) 113 assert self.app['applicants']['dp2012'] == 'New application' 104 114 105 115 def test_delEntry(self): 106 self.importer.delEntry(dict( reg_no='REG_NO_1'), self.site1)107 assert ' REG_NO_1' not in self.site1['jambdata'].keys()116 self.importer.delEntry(dict(code='dp2011'), self.app) 117 assert 'dp2011' not in self.app['applicants'].keys() 108 118 109 119 def test_import(self): 110 120 # Do a real import 111 # The following modules register important components for import 112 # Registers converters... 113 grok.testing.grok('waeup.sirp.utils.converters') 114 # Registers the Applicant factory... 115 grok.testing.grok('waeup.sirp.applicants.applicants') 121 self.importer.doImport( 122 self.csv_file, APPS_CONTAINER_HEADER_FIELDS) 123 #print self.workdir 124 #import pdb; pdb.set_trace() 116 125 117 # Create a fake site to store datasets in... 118 site = FakeSite() 119 # The site must have a 'jambdata' entry... 120 site['jambdata'] = dict() 121 # Set the fake site as 'current' site object... 122 setSite(site) 123 self.importer.doImport(self.csv_file, HEADER_FIELDS) 124 self.assertTrue(u'91100546DD' in site['jambdata'].keys()) 126 # This currently gives: 127 # 128 # ac_prefix,code,application_category,provider,prefix,year,--ERRORS-- 129 # APP,CODE1,app,base,app,2013,"conversion error: field application_category: <type 'exceptions.LookupError'> LookupError('app',) / conversion error: field provider: <type 'exceptions.TypeError'> TypeError('unhashable type',)" 130 # DPP,CODE2,dp,base,app,2013,"conversion error: field application_category: <type 'exceptions.LookupError'> LookupError('dp',) / conversion error: field provider: <type 'exceptions.TypeError'> TypeError('unhashable type',)" 131 self.assertTrue(u'CODE1' in self.app['applicants'].keys()) 125 132 126 133 def test_suite():
Note: See TracChangeset for help on using the changeset viewer.