source: ndgCommon/trunk/ndg/common/unittests/lib/testgranulite.py @ 5219

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/lib/testgranulite.py
Revision 5219, 9.0 KB checked in by cbyrom, 10 years ago (diff)

Various updates to test suite and a small fix to allow granulite to
process atom summaries properly as a string.

Line 
1'''
2A test class for the Granulite module.
3
4C Byrom Jun 2008
5Tessella
6'''
7import unittest, logging
8from ndg.common.src.lib.granulite import granulite
9import ndg.common.src.clients.xmldb.abstractxmldbatomclient as ac
10import ndg.common.unittests.testconstants as tc
11from ndg.common.unittests.testutils import testUtils as tu
12from ndg.common.src.models.Atom import Atom
13from ndg.common.src.models.vocabtermdata import VocabTermData as VTD
14import ndg.common.src.lib.utilities as utilities
15import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
16from ndg.common.src.models.ndgObject import ndgObject as no
17
18class testGranulite(unittest.TestCase):
19
20    def setUp(self):
21        '''
22        set up data used in the tests.
23        '''
24        self.utils = tu(tc.EXIST_DBCONFIG_FILE)
25        self.g = granulite(tc.VALID_TEST_GRANULITE, eXistClient = self.utils.ac)
26
27    def createGranuliteDEs(self):
28        '''
29        Create the DEs referenced in the granulite file
30        '''
31        inputs = self.g._granulite__getGranuliteDetails()
32        self.DEs = []
33        for de in inputs['data_entity_id']:
34            data = utilities.getTripleData(de)
35            id = data[0]
36            atom = self.utils.createAtom(tc.xmlString)
37            atom.atomTypeID = VTD.DE_TERM
38            atom.setDatasetID(id)
39            atom = self.utils.ac.createAtom(atom)
40            self.DEs.append(atom)
41
42    def deleteGranuliteDEs(self):
43        '''
44        Delete the DEs referenced in the granulite file
45        '''
46        for de in self.DEs:
47            self.utils.deleteDoc(de.getFullPath())
48           
49       
50    def testProcessGranulite(self):
51        atom = None
52        try:
53            # firstly, create the DEs referenced in the granulite
54            self.createGranuliteDEs()
55            atom = self.g.processGranulite(replaceAtom = True)
56            self.assertNotEqual(None, atom)
57            self.assertEqual(-9.140625, atom.minX)
58            self.assertEqual(-11.25, atom.maxY)
59            self.assertEqual(-3.1640625, atom.maxX)
60            self.assertEqual(-17.578125, atom.minY)
61            self.assertEqual(tc.VERTICAL_EXTENT, atom.ME.verticalExtent)
62            self.assertEqual(tc.SPATIAL_RESOLUTION, atom.ME.spatialResolution)
63
64        except Exception, e:
65            self.fail("An exception (%s) should not have been thrown" %(e.message or e))
66        finally:
67            if atom is not None:
68                self.utils.deleteDoc(atom.getFullPath())
69            self.deleteGranuliteDEs()
70           
71       
72       
73    def testDuplicateProcessGranulite(self):
74        atom = None
75        try:
76            # firstly, create the DEs referenced in the granulite
77            self.createGranuliteDEs()
78            atom = self.g.processGranulite(replaceAtom = False)
79            atom = self.g.processGranulite(replaceAtom = False, initialiseAtom = True)
80            self.fail("An exception should have been thrown")
81
82        except ac.DuplicateError, de:
83            self.assertTrue(de.message.startswith('An atom with the specified ID'))
84        except Exception, e:
85            self.fail("An exception (%s) should not have been thrown" %e)
86        finally:
87            if atom is not None:
88                self.utils.deleteDoc(atom.getFullPath())
89            self.deleteGranuliteDEs()
90       
91    def testProcessCSMLOrCDMLFile(self):
92        try:
93            atom = self.g.processCSMLOrCDMLFile()
94            self.assertNotEqual(None, atom)
95        except:
96            self.fail("An exception should not have been thrown")
97       
98       
99    def testGetSectionName(self):
100        sectionName = "parameters"
101        foundName = self.g._granulite__getSectionName(sectionName + "::")
102        self.assertEqual(sectionName, foundName)
103       
104    def testGetInvalidSectionName(self):
105        sectionName = "parameters"
106        foundName = self.g._granulite__getSectionName(sectionName)
107        self.assertEqual(None, foundName)
108       
109    def testInitialiseGranule(self):
110        self.g._granulite__initialiseGranule()
111        self.assertNotEqual(None, self.g._atom)
112       
113    def _initGranule(self):
114        self.g._granulite__initialiseGranule()
115        return self.g._granulite__getGranuliteDetails()
116       
117
118    def testGetGranuliteDetails(self):
119        details = self._initGranule()
120        self.assertEqual(tc.GRANULITE_TITLE, details['granule_info'][0].split('|')[2].strip())
121        self.assertEqual(tc.GRANULITE_TIME_AXIS, details['cdml_file'][0].split('|')[2].strip())
122        self.assertEqual(tc.GRANULITE_CDML_DATASETID, details['cdml_file'][0].split('|')[1].strip())
123
124    def testApplyCoreGranuliteDetails(self):
125        details = self._initGranule()
126        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
127        self.g._granulite__applyCoreGranuliteDetails(details)
128        self.assertEqual(tc.GRANULITE_TITLE, self.g._atom.title)
129        self.assertEqual(tc.GRANULITE_DATASETID, self.g._atom.datasetID)
130        self.assertEqual(tc.GRANULITE_PROVIDER_ID, self.g._atom.ME.providerID)
131
132    def testApplyGranuliteDetails(self):
133        details = self._initGranule()
134        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
135        self.g._granulite__applyGranuliteDetails(details)
136        self.assertEqual(tc.GRANULITE_LOGO, self.g._atom.getLogos()[0].href)
137        self.assertEquals(tc.GRANULITE_AUTHOR_NAME, self.g._atom.ME.responsibleParties[0].name)
138        self.assertEqual(-9.140625, self.g._atom.minX)
139        self.assertEqual(-11.25, self.g._atom.maxY)
140        self.assertEqual(-3.1640625, self.g._atom.maxX)
141        self.assertEqual(-17.578125, self.g._atom.minY)
142        self.assertEqual(self.g._atom.t1, tc.GRANULITE_T1)
143        self.assertEqual(self.g._atom.t2, tc.GRANULITE_T2)
144       
145
146    def testExtractSpatialData(self):
147        self._initGranule()
148        self.g._extractSpatialData("POLYGON((-8.26171875 -11.25,-9.140625 -17.40234375,-3.33984375 -17.578125,-3.1640625 -12.65625,-8.26171875 -11.25))")
149        self.assertEqual(-9.140625, self.g._atom.minX)
150        self.assertEqual(-11.25, self.g._atom.maxY)
151        self.assertEqual(-3.1640625, self.g._atom.maxX)
152        self.assertEqual(-17.578125, self.g._atom.minY)
153
154
155    def testUpdateDataEntity(self):
156        # firstly create a DE to use
157        try:
158            atom = Atom(VTD.DE_TERM)
159            atom.fromString(tc.xmlString)
160            atom.atomTypeID = VTD.DE_TERM
161            atom = self.utils.ac.createAtom(atom)
162   
163            # now set up the granulite
164            details = self._initGranule()
165            self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
166            self.g._granulite__applyCoreGranuliteDetails(details)
167           
168            # and now update DE
169            self.g._granulite__updateDataEntity(atom.datasetID, atom.ME.providerID)
170            atomPath = atom.getDefaultCollectionPath() + atom.atomName
171            self.assertEqual([], self.utils.ac.checkAtomSchemaCompliance(atomPath))
172   
173            xml = self.utils.ac.getNDGDoc(atom.ME.providerID, no.ATOM_DOC_TYPE,
174                                          tc.datasetID, 
175                                          targetCollection = dc.ATOM_COLLECTION_PATH)
176            updatedAtom = Atom(xmlString = xml)
177           
178            self.assertNotEqual(atom, updatedAtom)
179           
180            # there should be one extra link on the new atom
181            self.assertEqual(len(atom.relatedLinks) + 1, len(updatedAtom.relatedLinks))
182        except Exception, e:
183            self.fail("An exception shouldn't have been thrown")
184        finally:
185            self.utils.deleteDoc(atom.getFullPath())
186
187
188    def testAddGranuleToDataEntityRecords(self):
189        # firstly create a DE to use
190        atom = Atom(VTD.DE_TERM)
191        atom.fromString(tc.xmlString)
192        atom.atomTypeID = VTD.DE_TERM
193        atom = self.utils.ac.createAtom(atom)
194
195        # now set up the granulite
196        details = self._initGranule()
197        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
198        self.g._granulite__applyCoreGranuliteDetails(details)
199       
200        # and now update DE
201        self.g._dataEntityIDs = [atom.datasetID]
202        self.g._granulite__addGranuleToDataEntityRecords()
203        atomPath = atom.getDefaultCollectionPath() + atom.atomName
204        self.assertEqual([], self.utils.ac.checkAtomSchemaCompliance(atomPath))
205
206        xml = self.utils.ac.getNDGDoc(atom.ME.providerID, no.ATOM_DOC_TYPE,
207                                      tc.datasetID, 
208                                      targetCollection = dc.ATOM_COLLECTION_PATH)
209        updatedAtom = Atom(xmlString = xml)
210        self.assertNotEqual(atom, updatedAtom)
211       
212        # there should be one extra link on the new atom
213        self.assertEqual(len(atom.relatedLinks) + 1, len(updatedAtom.relatedLinks))
214        self.utils.deleteDoc(atomPath)
215       
216       
217if __name__ == '__main__':
218
219    unittest.main()
Note: See TracBrowser for help on using the repository browser.