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

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

Incomplete - # 22490: CEDA Observation Collection - Description
 http://team.ceda.ac.uk/trac/ceda/ticket/22490

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