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

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

Complete - # 22386: CEDA Observation - title
 http://team.ceda.ac.uk/trac/ceda/ticket/22386

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