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

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

Incomplete - # 22416: CEDA MOLES3 - search function - by title keyword
 http://team.ceda.ac.uk/trac/ceda/ticket/22416

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