1 | ''' |
---|
2 | Created on 8 Mar 2012 |
---|
3 | |
---|
4 | @author: mnagni |
---|
5 | ''' |
---|
6 | import unittest |
---|
7 | from testconfig import MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT |
---|
8 | from MolesManager.moles3epb import Moles3EPB |
---|
9 | from libs.commons_db import DbManager |
---|
10 | from unittest import TestCase |
---|
11 | import logging |
---|
12 | from logging import StreamHandler |
---|
13 | from test_utils import createObservationCollection, createObservation, createProject |
---|
14 | from testconfig import TESTMOLES3_DB_CONNECTION |
---|
15 | from ea_model.ceda_metadatamodel.ceda_observationcollection.ceda_observationcollection import CEDA_ObservationCollection |
---|
16 | from ea_model.moles3_4.observationcollection.mo_observationcollection import MO_ObservationCollection |
---|
17 | from ea_model.moles3_4.utilities.mo_responsiblepartyinfo import MO_ResponsiblePartyInfo |
---|
18 | from ea_model.ceda_metadatamodel.ceda_observation.ceda_observation import CEDA_Observation |
---|
19 | from ea_model.moles3_4.observation.mo_observation import MO_Observation |
---|
20 | from ea_model.moles3_4.utilities.mo_rolevalue import MO_RoleValue |
---|
21 | |
---|
22 | |
---|
23 | class 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 | |
---|
172 | if __name__ == "__main__": |
---|
173 | #import sys;sys.argv = ['', 'Test.testName'] |
---|
174 | unittest.main() |
---|