source: mauRepo/MolesManager/trunk/src/libs/migration/tests/moles3epbmothods.py @ 8295

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/mauRepo/MolesManager/trunk/src/libs/migration/tests/moles3epbmothods.py@8295
Revision 8295, 7.2 KB checked in by mnagni, 7 years ago (diff)

added a method on moles3epb to retrieve a CEDA_Observation author

Line 
1'''
2Created on 8 Mar 2012
3
4@author: mnagni
5'''
6import unittest
7from testconfig import MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT
8from MolesManager.moles3epb import Moles3EPB
9from libs.commons_db import DbManager
10from unittest import TestCase
11import logging
12from logging import StreamHandler
13from test_utils import createObservationCollection, createObservation, createProject
14from testconfig import TESTMOLES3_DB_CONNECTION
15from ea_model.ceda_metadatamodel.ceda_observationcollection.ceda_observationcollection import CEDA_ObservationCollection
16from ea_model.moles3_4.observationcollection.mo_observationcollection import MO_ObservationCollection
17from ea_model.moles3_4.utilities.mo_responsiblepartyinfo import MO_ResponsiblePartyInfo
18from ea_model.ceda_metadatamodel.ceda_observation.ceda_observation import CEDA_Observation
19from ea_model.moles3_4.observation.mo_observation import MO_Observation
20from ea_model.moles3_4.utilities.mo_rolevalue import MO_RoleValue
21
22
23class Moles3EPBTest(TestCase):
24
25    def setUp(self):   
26        Moles3EPBTest.molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT)
27        #Moles3EPBTest.molesDB = DbManager(TESTMOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT)
28        Moles3EPB.overrrideDBManager(Moles3EPBTest.molesDB)
29        self._initSearchIndexes(Moles3EPBTest.molesDB)       
30        self.logging = logging.getLogger('Moles3EPBTest')
31        self.logging.addHandler(StreamHandler())
32        self.logging.setLevel(logging.DEBUG)
33
34    def tearDown(self):
35        pass
36        #self._dropAllTables()
37
38    def runTest(self):
39        #self.checkExtractCollectionIdentifierByTitle()   
40        #self.checkExtractObservationByTitleKeywords()
41        #self.checkExtractProjectObservationCollections()
42        #self.checkExtractObservationCollectionsForObservation()
43        self.checkObservationAuthors()
44
45    def checkObservationAuthors(self):
46        session = Moles3EPB.getNewMolesSession();
47        coll = Moles3EPB.observationAuthor(1, session)
48        colls = coll.all()
49        print colls
50
51    def checkExtractObservationCollectionsForObservation(self):
52        session = Moles3EPB.getNewMolesSession();
53        #coll = session.query(CEDA_ObservationCollection). \
54        #    join(MO_ObservationCollection).join(MO_ObservationCollection.member)
55        coll = session.query(CEDA_ObservationCollection)
56        colls = coll.all()
57        print colls
58
59    def checkExtractCollectionIdentifierByTitle(self):
60        session = Moles3EPB.getNewMolesSession();
61        observationCollection = createObservationCollection()
62       
63        self.logging.info('Stores an new CEDA_ObservationCollection')
64        Moles3EPB.addCedaObject(observationCollection, session)
65        session.commit()
66
67        my_identifier = Moles3EPB.extractCollectionIdentifierByTitle('test_title', session)           
68        self.assertTrue(my_identifier.count() > 0, "Cannot find MD_Identifier")
69        my_identifier = Moles3EPB.extractCollectionIdentifierByTitle('fake_title', session)       
70        self.assertTrue(my_identifier.count() == 0, "Cannot find MD_Identifier")           
71       
72        self.logging.info('Deletes it')
73        session.delete(observationCollection)
74        session.commit()
75
76    def checkExtractObservationByTitleKeywords(self):
77        session = Moles3EPB.getNewMolesSession();
78        observation = createObservation()
79       
80        self.logging.info('Stores an new CEDA_Observation')
81        Moles3EPB.addCedaObject(observation, session)
82        session.commit()
83       
84        obs = Moles3EPB.extractObservationByTitleKeywords('test_code', session)
85        self.assertTrue(obs.count() > 1, "Cannot find a CEDA_Observation")
86        obs = Moles3EPB.extractObservationByTitleKeywords('dummy_code', session)
87        self.assertTrue(obs.count() == 0, "Should NOT find a CEDA_Observation")
88        session.delete(observation)
89        session.commit()           
90
91    def checkExtractProjectObservationCollections(self):
92        session = Moles3EPB.getNewMolesSession();
93
94        p_1 = createProject()
95        p_2 = createProject()
96        p_3 = createProject()       
97        p_4 = createProject()   
98        self._create([p_1, p_2, p_3, p_4], session) 
99       
100        o_1 = createObservation()
101        o_1.inSupportOf = p_1               
102        o_2 = createObservation()
103        o_2.inSupportOf = p_2       
104        o_3 = createObservation()               
105        o_3.inSupportOf = p_3
106        self._create([o_1, o_2, o_3], session)
107
108        oc_1 = createObservationCollection()
109        oc_2 = createObservationCollection()
110        oc_3 = createObservationCollection()       
111        self._create([oc_1, oc_2, oc_3], session)
112        oc_1.member.append(o_1)
113        session.commit()
114       
115        oc_2.member.append(o_1)
116        oc_2.member.append(o_2)
117        session.commit()
118       
119        oc_3.member.append(o_2)
120        oc_3.member.append(o_3)       
121        session.commit()
122
123        objects = [p_1, p_2, p_3, p_4, o_1, o_2, o_3, oc_1, oc_2, oc_3]
124
125
126               
127
128        try:           
129            res = Moles3EPB.extractProjectObservationCollections(p_1, session)
130            print res.count()
131            #for item in res:
132            #    print item
133            #self.assertTrue(res.count() == 2, "Cannot find a CEDA_Observationcollection")
134            #res = Moles3EPB.extractProjectObservationCollections(p_2, session)
135            #self.assertTrue(res.count() == 2, "Cannot find a CEDA_Observationcollection")
136            #res = Moles3EPB.extractProjectObservationCollections(p_3, session)
137            #self.assertTrue(res.count() == 1, "Cannot find a CEDA_Observationcollection")                       
138            #res = Moles3EPB.extractProjectObservationCollections(p_4, session)
139            #self.assertTrue(res.count() == 0, "Cannot find a CEDA_Observationcollection")           
140        except Exception as e:
141            print e
142            self._delete(objects, session)
143            self.assertTrue(True==False, "Error!")
144        self._delete(objects, session)               
145
146                   
147    def _initSearchIndexes(self, db_manager):
148        #To Be Done - CHECK IF THE COLUMN ALREADY EXISTS!
149        # We don't want sqlalchemy to know about this column so we add it externally.
150        try:
151            db_manager._engine.execute("alter table md_identifier add column code_search_vector tsvector")                 
152
153            # This indexes the tsvector column
154            db_manager._engine.execute("create index md_identifier_code_search_index on md_identifier using gin(code_search_vector)")
155           
156            # This sets up the trigger that keeps the tsvector column up to date.
157            db_manager._engine.execute("create trigger md_identifier_code_search_update before update or insert on md_identifier \
158                for each row execute procedure tsvector_update_trigger('code_search_vector', 'pg_catalog.english', 'code')")           
159        except Exception as e:
160            pass
161
162    def _create(self, objects, session):
163        for item in objects:
164            Moles3EPB.addCedaObject(item, session)
165        session.commit()
166
167    def _delete(self, objects, session):
168        for item in objects:
169            session.delete(item)
170        session.commit()
171
172if __name__ == "__main__":
173    #import sys;sys.argv = ['', 'Test.testName']
174    unittest.main()
Note: See TracBrowser for help on using the repository browser.