Changeset 8217
- Timestamp:
- 10/04/12 21:10:18 (9 years ago)
- Location:
- mauRepo/MolesManager/trunk/src
- Files:
-
- 7 edited
Legend:
- Unmodified
- Added
- Removed
-
mauRepo/MolesManager/trunk/src/MolesManager/moles3epb.py
r8205 r8217 14 14 from ea_model.moles3_4.observationcollection.mo_observationcollection import MO_ObservationCollection 15 15 from ea_model.moles3_4.observation.mo_observation import MO_Observation 16 from ea_model.ceda_metadatamodel.ceda_project.ceda_project import CEDA_Project 16 17 17 18 #molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT) … … 89 90 def extractCollectionIdentifierByTitle(self, i_title, session = None): 90 91 """ 91 Loooks for an MD_Identifier from a CEDA_ObservationCollection contains a specific title (observation.identifier.code)92 Searches for an MD_Identifier from a CEDA_ObservationCollection contains a specific title (observation.identifier.code) 92 93 @param i_title: the CEDA_ObservationCollection.identifier.title value to search for 93 94 @return: a tuple containing a CEDA_ObservationCollection and the CEDA_ObservationCollection.idenfitier element having the title … … 97 98 join(MO_ObservationCollection).join(MO_ObservationCollection.identifier). \ 98 99 join(MD_Identifier.authority).filter(CI_Citation.title.like('%' + i_title + '%')) 99 100 101 @classmethod 102 def extractObservationsForProject(self, project, session = None): 103 """ 104 Searches for the CEDA_Observation associated with a CEDA_Project 105 @param project: a CEDA_Project instance 106 @return: a tuple containing the associated CEDA_Observation 107 """ 108 intSession = Moles3EPB._getSession(session) 109 return intSession.query(CEDA_Observation). \ 110 join(CEDA_Observation, MO_Observation.inSupportOf).filter(CEDA_Project.id == project.id) 111 112 @classmethod 113 def extractProjectObservationCollections(self, project, session = None): 114 """ 115 Searches for the Observation_Collections associated with a CEDA_Project 116 @param project: a CEDA_Project instance 117 @return: a tuple containing the associated CEDA_ObservationCollection 118 """ 119 intSession = Moles3EPB._getSession(session) 120 mo_obs = intSession.query(MO_Observation).join(CEDA_Project).filter(CEDA_Project.id == project.id).subquery() 121 obsers = intSession.query(CEDA_Observation).join(mo_obs, CEDA_Observation.id == mo_obs.c.id).one() 122 #print "obsers: " + str(intSession.query(CEDA_Observation).join(mo_obs, CEDA_Observation.id == mo_obs.c.id).count()) 123 124 cos = intSession.query(CEDA_ObservationCollection).all() 125 co = intSession.query(MO_ObservationCollection).join(MO_ObservationCollection.member).filter(MO_ObservationCollection.member.contains(obsers)) 126 print co 127 print co.all() 128 129 observations = intSession.query(MO_ObservationCollection).join(CEDA_Observation). \ 130 filter(obsers.any(CEDA_Observation.id==obsers.c.id)) 131 print "observation:" + str(observations.count()) 132 return observations 133 100 134 @classmethod 101 135 def addObservationToObservationCollection(self, observationCollection, observation, session = None, commit = False): -
mauRepo/MolesManager/trunk/src/MolesManager/molesSessionMiddleware.py
r8211 r8217 38 38 pass 39 39 40 41 40 def _getNewMolesSession(self): 41 session = Moles3EPB.getNewMolesSession() 42 MOLES3_DB_SCRIPT.attachEvents(session) 43 return session 42 44 43 45 def _getMolesSession(self): … … 51 53 52 54 self._initSearchIndexes(molesDB) 53 54 return Moles3EPB.getNewMolesSession() 55 else: 56 return Moles3EPB.getNewMolesSession() 55 56 return self._getNewMolesSession() 57 57 58 58 def _migration(self, runMigration = RUN_MIGRATION): -
mauRepo/MolesManager/trunk/src/libs/migration/processor/commons.py
r8210 r8217 616 616 return ci_party 617 617 618 def createCI_Organization(name = None, contactInfo = None): 619 ''' 620 Creates a new CI_Individual instance 621 @param name: a String 622 @param contactInfo: an array of CI_Contact 623 ''' 624 ci_party = CI_Organisation() 625 if name: 626 ci_party.name = name 627 if contactInfo: 628 ci_party.contactInfo = contactInfo 629 return ci_party 630 618 631 def createMO_Organization(name = None, contactInfo = None): 619 632 ''' -
mauRepo/MolesManager/trunk/src/libs/migration/tests/moles3epbmothods.py
r8205 r8217 11 11 import logging 12 12 from logging import StreamHandler 13 from test_utils import createObservationCollection, createObservation 14 from sqlalchemy.dialects import postgres13 from test_utils import createObservationCollection, createObservation, createProject 14 from ea_model.moles3_4.observationcollection.mo_observationcollection import MO_ObservationCollection 15 15 16 16 … … 30 30 31 31 def runTest(self): 32 self.checkExtractCollectionIdentifierByTitle() 33 self.checkExtractObservationByTitleKeywords() 32 #self.checkExtractCollectionIdentifierByTitle() 33 #self.checkExtractObservationByTitleKeywords() 34 self.checkExtractProjectObservationCollections() 34 35 35 36 def checkExtractCollectionIdentifierByTitle(self): … … 61 62 self.assertTrue(obs.count() > 1, "Cannot find a CEDA_Observation") 62 63 obs = Moles3EPB.extractObservationByTitleKeywords('dummy_code', session) 63 self.assertTrue(obs.count() == 0, "Should NOT find a CEDA_Observation") 64 self.assertTrue(obs.count() == 0, "Should NOT find a CEDA_Observation") 65 session.delete(observation) 66 session.commit() 64 67 68 def checkExtractProjectObservationCollections(self): 69 session = Moles3EPB.getNewMolesSession(); 70 71 p_1 = createProject() 72 p_2 = createProject() 73 p_3 = createProject() 74 p_4 = createProject() 75 self._create([p_1, p_2, p_3, p_4], session) 76 77 o_1 = createObservation() 78 o_1.inSupportOf = p_1 79 o_2 = createObservation() 80 o_2.inSupportOf = p_2 81 o_3 = createObservation() 82 o_3.inSupportOf = p_3 83 self._create([o_1, o_2, o_3], session) 84 85 oc_1 = createObservationCollection() 86 oc_2 = createObservationCollection() 87 oc_3 = createObservationCollection() 88 self._create([oc_1, oc_2, oc_3], session) 89 oc_1.member.append(o_1) 90 session.commit() 91 92 oc_2.member.append(o_1) 93 oc_2.member.append(o_2) 94 session.commit() 95 96 oc_3.member.append(o_2) 97 oc_3.member.append(o_3) 98 session.commit() 99 100 objects = [p_1, p_2, p_3, p_4, o_1, o_2, o_3, oc_1, oc_2, oc_3] 101 102 103 104 105 try: 106 res = Moles3EPB.extractProjectObservationCollections(p_1, session) 107 print res.count() 108 #for item in res: 109 # print item 110 #self.assertTrue(res.count() == 2, "Cannot find a CEDA_Observationcollection") 111 #res = Moles3EPB.extractProjectObservationCollections(p_2, session) 112 #self.assertTrue(res.count() == 2, "Cannot find a CEDA_Observationcollection") 113 #res = Moles3EPB.extractProjectObservationCollections(p_3, session) 114 #self.assertTrue(res.count() == 1, "Cannot find a CEDA_Observationcollection") 115 #res = Moles3EPB.extractProjectObservationCollections(p_4, session) 116 #self.assertTrue(res.count() == 0, "Cannot find a CEDA_Observationcollection") 117 except Exception as e: 118 print e 119 self._delete(objects, session) 120 self.assertTrue(True==False, "Error!") 121 self._delete(objects, session) 122 123 65 124 def _initSearchIndexes(self, db_manager): 66 125 #To Be Done - CHECK IF THE COLUMN ALREADY EXISTS! … … 78 137 pass 79 138 139 def _create(self, objects, session): 140 for item in objects: 141 Moles3EPB.addCedaObject(item, session) 142 session.commit() 80 143 144 def _delete(self, objects, session): 145 for item in objects: 146 session.delete(item) 147 session.commit() 81 148 82 149 if __name__ == "__main__": -
mauRepo/MolesManager/trunk/src/libs/migration/tests/moles3epbtests.py
r8208 r8217 5 5 ''' 6 6 import unittest 7 from testconfig import MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT 7 from testconfig import MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT, EVENTS_DB 8 8 from MolesManager.moles3epb import Moles3EPB 9 9 from libs.commons_db import DbManager … … 16 16 from libs.migration.processor.check.check import moles3Append 17 17 from libs.migration.processor.commons import createCI_Date, createDateTime,\ 18 createDate, createTM_Position, createTM_Instant, isoDateTimeStringToTimeDate18 createDate, createTM_Position, createTM_Instant, createCI_Individual 19 19 from ea_model.iso_19115_2006_metadata_corrigendum.citation_and_responsible_party_information.ci_datetypecode import CI_DateTypeCode 20 20 from ea_model.iso_19115_2006_metadata_corrigendum.citation_and_responsible_party_information.ci_date import CI_Date … … 22 22 from ea_model.iso_19108_2006_temporal_schema.temporal_reference_system.tm_position import TM_Position 23 23 from test_utils import createObservationCollection,\ 24 createProject 25 from ea_model.ceda_metadatamodel.ceda_project.ceda_project import CEDA_Project 24 createProject, createCI_Responsibility 25 from ea_model.ceda_metadatamodel.ceda_project.ceda_project import CEDA_Project 26 26 27 27 … … 30 30 def setUp(self): 31 31 #self._dropAllTables() 32 Moles3EPBTest.molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT) 32 Moles3EPBTest.molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT) 33 33 Moles3EPB.overrrideDBManager(Moles3EPBTest.molesDB) 34 34 self.logging = logging.getLogger('Moles3EPBTest') … … 41 41 42 42 def runTest(self): 43 self.checkEmptyObservationCollection() 44 self.checkEmptyObservation() 45 self.checkObservation() 46 self.checkObservationCollection() 47 self.checkCI_Date() 48 self.checkTM_Instant() 49 self.checkTM_Position() 43 #self.checkEmptyObservationCollection() 44 #self.checkEmptyObservation() 45 #self.checkObservation() 46 #self.checkObservationCollection() 47 #self.checkCI_Date() 48 #self.checkTM_Instant() 49 #self.checkTM_Position() 50 self.checkCI_Responsibility() 50 51 51 52 def checkEmptyObservationCollection(self): … … 205 206 session.commit() 206 207 208 def checkCI_Responsibility(self): 209 session = Moles3EPB.getNewMolesSession(); 210 EVENTS_DB(session) 211 responsibility = createCI_Responsibility() 212 self.logging.info('Stores an empty new CEDA_Project') 213 Moles3EPB.addCedaObject(responsibility, session) 214 session.commit() 215 session.expunge(responsibility) 216 217 ci_individual_2 = createCI_Individual('individual_2') 218 responsibility.party.append(ci_individual_2) 219 session.merge(responsibility, cascade="merge") 220 session.commit() 221 222 for item in responsibility.party: 223 if item.name == 'individual_1': 224 responsibility.party.remove(item) 225 226 session.merge(responsibility) 227 session.commit() 228 print responsibility 229 session.refresh(responsibility) 230 session.delete(responsibility) 231 session.commit() 232 207 233 208 234 def _createObservation(self): -
mauRepo/MolesManager/trunk/src/libs/migration/tests/test_utils.py
r8208 r8217 6 6 from ea_model.ceda_metadatamodel.ceda_observationcollection.ceda_observationcollection import CEDA_ObservationCollection 7 7 from libs.migration.processor.commons import createMD_Identifier,\ 8 createCI_Citation 8 createCI_Citation, createCI_Individual, createCI_Organization 9 9 from ea_model.ceda_metadatamodel.ceda_observation.ceda_observation import CEDA_Observation 10 10 from ea_model.ceda_metadatamodel.ceda_project.ceda_project import CEDA_Project 11 from ea_model.upcomingiso.ci_responsibility import CI_Responsibility 12 from ea_model.iso_19115_2006_metadata_corrigendum.citation_and_responsible_party_information import ci_rolecode 13 from ea_model.upcomingiso.ci_individual import CI_Individual 11 14 12 15 def createObservationCollection(): … … 38 41 return observation 39 42 43 def createCI_Responsibility(): 44 ''' 45 Creates a CI_Responsibility with the following data: 46 ci_responsibilty.role = 'ci_rolecode.CI_RoleCode.cl_author 47 ci_responsibilty.party[0] 48 CI_Individual.name = 'individual_1' 49 ci_responsibilty.party[1] 50 CI_Organization.name = 'organization_1' 51 ''' 52 ci_responsibilty = CI_Responsibility() 53 ci_responsibilty.role = ci_rolecode.CI_RoleCode.cl_author 54 ci_individual_1 = createCI_Individual('individual_1') 55 ci_responsibilty.party.append(ci_individual_1) 56 ci_organization_1 = createCI_Organization('organization_1') 57 ci_responsibilty.party.append(ci_organization_1) 58 return ci_responsibilty 59 40 60 def createProject(): 41 61 ''' -
mauRepo/MolesManager/trunk/src/libs/migration/tests/testconfig.py
r8147 r8217 10 10 MIGRATION_DB_SCRIPT = doMigration 11 11 12 from sqlTables import doTables as doMoles 13 MOLES3_DB_SCRIPT = doMoles 12 13 from sqlTables import doTables as doMoles, attachEvents 14 MOLES3_DB_SCRIPT = doMoles 15 EVENTS_DB = attachEvents
Note: See TracChangeset
for help on using the changeset viewer.