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

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/mauRepo/MolesManager/trunk/src/libs/migration/tests/moles3epbtests.py@8475
Revision 8409, 10.4 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, EVENTS_DB
8from MolesManager.moles3epb import Moles3EPB
9from libs.commons_db import DbManager
10from unittest import TestCase
11import logging, datetime
12from logging import StreamHandler
13from sqlalchemy.exc import IntegrityError
14from libs.migration.processor.check.check import moles3Append
15from libs.migration.processor.commons import createCI_Date, createDateTime,\
16    createDate, createTM_Position, createTM_Instant, createCI_Individual
17from ea_model.iso_19115_2006_metadata_corrigendum.citation_and_responsible_party_information.ci_datetypecode import CI_DateTypeCode
18from ea_model.iso_19115_2006_metadata_corrigendum.citation_and_responsible_party_information.ci_date import CI_Date
19from ea_model.iso_19108_2006_temporal_schema.temporal_objects.tm_instant import TM_Instant
20from ea_model.iso_19108_2006_temporal_schema.temporal_reference_system.tm_position import TM_Position
21from test_utils import createObservationCollection,\
22    createProject, createCI_Responsibility 
23
24
25class Moles3EPBTest(TestCase):
26
27    def setUp(self):
28        #self._dropAllTables()     
29        Moles3EPBTest.molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT)       
30        Moles3EPB.overrrideDBManager(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.checkEmptyObservationCollection()
41        #self.checkEmptyObservation()
42        #self.checkObservation()
43        #self.checkObservationCollection()
44        #self.checkCI_Date()
45        #self.checkTM_Instant()
46        #self.checkTM_Position()
47        self.checkCI_Responsibility() 
48
49    def checkEmptyObservationCollection(self):
50        session = Moles3EPB.getNewMolesSession();
51        observationCollection = createObservationCollection()
52       
53        self.logging.info('Stores an empty new CEDA_ObservationCollection')
54        Moles3EPB.persistInstance(observationCollection, session)
55        session.commit()
56       
57        self.logging.info('Tries to retrieve it')
58        self.assertTrue(hasattr(observationCollection, 'id'), "Cannot create  ObservationCollection")       
59        res = Moles3EPB.search(CEDA_ObservationCollection, observationCollection.id, session)
60        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
61       
62        self.logging.info('Deletes it')
63        session.delete(observationCollection)
64        session.commit()
65       
66    def checkEmptyObservation(self):
67        session = Moles3EPB.getNewMolesSession();
68        observation = CEDA_Observation()
69       
70        self.logging.info('Tries to stores an empty new %s' % (observation.__class__.__name__))
71        Moles3EPB.persistInstance(observation, session)
72        self.assertRaises(IntegrityError, session.commit)
73        self.logging.info('Catches that Ceda_Observation.dataLineage is missing')
74        session.rollback()
75       
76        self.logging.info('Adds the dataLineage and tries to store it again')
77        observation.dataLineage = "aNewDataLineage"       
78        Moles3EPB.persistInstance(observation, session)       
79        session.commit()       
80       
81        self.logging.info('Tries to retrieve it')
82        self.assertTrue(hasattr(observation, 'id'), 'Cannot create %s' % (observation.__class__.__name__))       
83        res = Moles3EPB.search(CEDA_Observation, observation.id, session)
84        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
85       
86        self.logging.info('Deletes it')
87        session.delete(observation)
88        session.commit()
89
90    def checkObservationCollection(self):
91        session = Moles3EPB.getNewMolesSession();
92        observationCollection = createObservationCollection()
93       
94        self.logging.info('Stores an empty new CEDA_ObservationCollection')
95        Moles3EPB.persistInstance(observationCollection, session)
96        session.commit()
97       
98        self.logging.info('Tries to retrieve it')
99        self.assertTrue(hasattr(observationCollection, 'id'), "Cannot create  ObservationCollection")       
100        res = Moles3EPB.search(CEDA_ObservationCollection, observationCollection.id, session)
101        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
102       
103        self.logging.info('Adds two CEDA_Observations')
104        firstObservation = self._createObservation()
105        observationCollection.member.append(firstObservation)
106        secondObservation = self._createObservation()
107        observationCollection.member.append(secondObservation)     
108        session.commit()
109       
110        self.assertTrue(Moles3EPB.observationCollectionHasObservation(observationCollection.id, firstObservation.id, session), "The collection should contains the given observation")
111       
112        self.logging.info('Deletes it')
113        session.delete(observationCollection)
114        session.commit()
115
116    def checkObservation(self):
117        session = Moles3EPB.getNewMolesSession();
118        observation = self._createObservation()
119       
120        self.logging.info('Tries to stores a new %s' % (observation.__class__.__name__))
121        Moles3EPB.persistInstance(observation, session)       
122        session.commit()       
123       
124        self.logging.info('Tries to retrieve it')
125        self.assertTrue(hasattr(observation, 'id'), 'Cannot create %s' % (observation.__class__.__name__))       
126        res = Moles3EPB.search(CEDA_Observation, observation.id, session)
127        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
128       
129        self.logging.info('Deletes it')
130        session.delete(observation)
131        session.commit()
132
133    def checkCI_Date(self):
134        session = Moles3EPB.getNewMolesSession();
135        py_date = datetime.date(2011, 4, 1)
136        dt = createDate(py_date)
137        ci_date = createCI_Date(CI_DateTypeCode.cl_creation, dt)
138       
139        self.logging.info('Stores an empty new CEDA_ObservationCollection')
140        Moles3EPB.persistInstance(ci_date, session)
141        session.commit()
142       
143        self.logging.info('Tries to retrieve it')
144        self.assertTrue(hasattr(ci_date, 'id'), "Cannot create  CI_Date")       
145        res = Moles3EPB.search(CI_Date, ci_date.id, session)
146        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
147       
148        self.logging.info('Deletes it')
149        session.delete(ci_date)
150        session.commit()
151
152    def checkTM_Position(self):
153        session = Moles3EPB.getNewMolesSession();
154        tm_position = self._createTM_Position()
155       
156        self.logging.info('Stores an empty new TM_Position')
157        Moles3EPB.persistInstance(tm_position, session)
158        session.commit()
159       
160        self.logging.info('Tries to retrieve it')
161        self.assertTrue(hasattr(tm_position, 'id'), "Cannot create  TM_Position")       
162        res = Moles3EPB.search(TM_Position, tm_position.id, session)
163        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
164       
165        self.logging.info('Deletes it')
166        session.delete(tm_position)
167        session.commit()       
168
169
170    def checkTM_Instant(self):
171        session = Moles3EPB.getNewMolesSession();       
172        tm_instant = self._createTM_Instant()
173
174        self.logging.info('Stores an empty new TM_Instant')
175        Moles3EPB.persistInstance(tm_instant, session)
176        session.commit()
177       
178        self.logging.info('Tries to retrieve it')
179        self.assertTrue(hasattr(tm_instant, 'id'), "Cannot create  TM_Instant")       
180        res = Moles3EPB.search(TM_Instant, tm_instant.id, session)
181        self.assertNotEqual(res, None, "Cannot retrieve the stored instance")
182       
183        self.logging.info('Deletes it')
184        session.delete(tm_instant)
185        session.commit()
186
187    def checkCEDA_Project(self):
188        session = Moles3EPB.getNewMolesSession();
189        project = createProject()
190   
191        self.logging.info('Stores an empty new CEDA_Project')
192        Moles3EPB.persistInstance(project, session)
193        session.commit()
194       
195        self.logging.info('Tries to retrieve it')
196        self.assertTrue(hasattr(project, 'id'), "Cannot create  TM_Instant")       
197        res = Moles3EPB.search(CEDA_Project, project.id, session)
198        for item in res:               
199            self.assertTrue(item.abstract == 'test_abstract', "Cannot retrieve the abstract from stored project")
200       
201        self.logging.info('Deletes it')
202        session.delete(project)
203        session.commit()
204   
205    def checkCI_Responsibility(self):
206        session = Moles3EPB.getNewMolesSession();
207        EVENTS_DB(session)       
208        responsibility = createCI_Responsibility()
209        self.logging.info('Stores an empty new CEDA_Project')
210        Moles3EPB.persistInstance(responsibility, session)
211        session.commit()
212        session.expunge(responsibility)
213       
214        ci_individual_2 = createCI_Individual('individual_2')
215        responsibility.party.append(ci_individual_2)
216        session.merge(responsibility, cascade="merge")       
217        session.commit()       
218       
219        for item in responsibility.party:
220            if item.name == 'individual_1':
221                responsibility.party.remove(item)
222               
223        session.merge(responsibility)
224        session.commit()       
225        print responsibility
226        session.refresh(responsibility)
227        session.delete(responsibility)
228        session.commit()       
229       
230
231    def _createObservation(self):
232        observation = CEDA_Observation()
233        observation.dataLineage = "dataLineage"
234        return observation
235   
236    def _createDateTime(self):
237        py_date = datetime.datetime(2011, 4, 1, 00, 00, 00)
238        return createDateTime(py_date)   
239   
240    def _createCI_Date(self):
241        py_date = datetime.date(2011, 4, 1)
242        dt = createDate(py_date)
243        return createCI_Date(CI_DateTypeCode.cl_creation, dt)
244   
245    def _createTM_Position(self):
246        newDateTime = self._createDateTime()               
247        return createTM_Position(dateTime8601 = newDateTime)
248
249    def _createTM_Instant(self):
250        tm_position = self._createTM_Position()       
251        return createTM_Instant(tm_position)
252
253if __name__ == "__main__":
254    #import sys;sys.argv = ['', 'Test.testName']
255    unittest.main()
Note: See TracBrowser for help on using the repository browser.