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

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

Add better error handling + add general keyword catcher to avoid
issues with slight differences between the search clients + extend test data and test suites - handling proxies better.

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       
51    def testProcessGranulite(self):
52        atom = None
53        try:
54            # firstly, create the DEs referenced in the granulite
55            self.createGranuliteDEs()
56            atom = self.g.processGranulite(replaceAtom = True)
57            self.assertNotEqual(None, atom)
58            self.assertEqual(-9.140625, atom.minX)
59            self.assertEqual(-11.25, atom.maxY)
60            self.assertEqual(-3.1640625, atom.maxX)
61            self.assertEqual(-17.578125, atom.minY)
62
63        except Exception, e:
64            self.fail("An exception should not have been thrown")
65        finally:
66            if atom is not None:
67                self.utils.deleteDoc(atom.getFullPath())
68            self.deleteGranuliteDEs()
69           
70       
71       
72    def testDuplicateProcessGranulite(self):
73        atom = None
74        try:
75            # firstly, create the DEs referenced in the granulite
76            self.createGranuliteDEs()
77            atom = self.g.processGranulite(replaceAtom = False)
78            atom = self.g.processGranulite(replaceAtom = False, initialiseAtom = True)
79            self.fail("An exception should not have been thrown")
80
81        except ac.DuplicateError, de:
82            self.assertTrue(de.message.startswith('An atom with the specified ID'))
83        except Exception, e:
84            self.fail("An exception should not have been thrown")
85        finally:
86            if atom is not None:
87                self.utils.deleteDoc(atom.getFullPath())
88            self.deleteGranuliteDEs()
89       
90    def testProcessCSMLOrCDMLFile(self):
91        try:
92            atom = self.g.processCSMLOrCDMLFile()
93            self.assertNotEqual(None, atom)
94        except:
95            self.fail("An exception should not have been thrown")
96       
97       
98    def testGetSectionName(self):
99        sectionName = "parameters"
100        foundName = self.g._granulite__getSectionName(sectionName + "::")
101        self.assertEqual(sectionName, foundName)
102       
103    def testGetInvalidSectionName(self):
104        sectionName = "parameters"
105        foundName = self.g._granulite__getSectionName(sectionName)
106        self.assertEqual(None, foundName)
107       
108    def testInitialiseGranule(self):
109        self.g._granulite__initialiseGranule()
110        self.assertNotEqual(None, self.g._atom)
111       
112    def _initGranule(self):
113        self.g._granulite__initialiseGranule()
114        return self.g._granulite__getGranuliteDetails()
115       
116
117    def testGetGranuliteDetails(self):
118        details = self._initGranule()
119        self.assertEqual(tc.GRANULITE_TITLE, details['granule_info'][0].split('|')[2].strip())
120        self.assertEqual(tc.GRANULITE_TIME_AXIS, details['cdml_file'][0].split('|')[2].strip())
121        self.assertEqual(tc.GRANULITE_CDML_DATASETID, details['cdml_file'][0].split('|')[1].strip())
122
123    def testApplyCoreGranuliteDetails(self):
124        details = self._initGranule()
125        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
126        self.g._granulite__applyCoreGranuliteDetails(details)
127        self.assertEqual(tc.GRANULITE_TITLE, self.g._atom.title)
128        self.assertEqual(tc.GRANULITE_DATASETID, self.g._atom.datasetID)
129        self.assertEqual(tc.GRANULITE_PROVIDER_ID, self.g._atom.ME.providerID)
130
131    def testApplyGranuliteDetails(self):
132        details = self._initGranule()
133        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
134        self.g._granulite__applyGranuliteDetails(details)
135        self.assertEqual(tc.GRANULITE_LOGO, self.g._atom.getLogos()[0].href)
136        self.assertEquals(tc.GRANULITE_AUTHOR_NAME, self.g._atom.ME.responsibleParties[0].name)
137        self.assertEqual(self.g._atom.minX, '-8.26171875')
138        self.assertEqual(self.g._atom.minY, '-11.25')
139        self.assertEqual(self.g._atom.maxX, '-3.33984375')
140        self.assertEqual(self.g._atom.maxY, '-17.578125')
141        self.assertEqual(self.g._atom.t1, tc.GRANULITE_T1)
142        self.assertEqual(self.g._atom.t2, tc.GRANULITE_T2)
143       
144
145    def testExtractSpatialData(self):
146        self._initGranule()
147        self.g._extractSpatialData("POLYGON((-8.26171875 -11.25,-9.140625 -17.40234375,-3.33984375 -17.578125,-3.1640625 -12.65625,-8.26171875 -11.25))")
148        self.assertEqual(self.g._atom.minX, '-8.26171875')
149        self.assertEqual(self.g._atom.minY, '-11.25')
150        self.assertEqual(self.g._atom.maxX, '-3.33984375')
151        self.assertEqual(self.g._atom.maxY, '-17.578125')
152
153
154    def testUpdateDataEntity(self):
155        # firstly create a DE to use
156        try:
157            atom = Atom(VTD.DE_TERM)
158            atom.fromString(tc.xmlString)
159            atom.atomTypeID = VTD.DE_TERM
160            atom = self.utils.ac.createAtom(atom)
161   
162            # now set up the granulite
163            details = self._initGranule()
164            self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
165            self.g._granulite__applyCoreGranuliteDetails(details)
166           
167            # and now update DE
168            self.g._granulite__updateDataEntity(atom.datasetID, atom.ME.providerID)
169            atomPath = atom.getDefaultCollectionPath() + atom.atomName
170            self.assertEqual([], self.utils.ac.checkAtomSchemaCompliance(atomPath))
171   
172            xml = self.utils.ac.getNDGDoc(atom.ME.providerID, no.ATOM_DOC_TYPE,
173                                          tc.datasetID, 
174                                          targetCollection = dc.ATOM_COLLECTION_PATH)
175            updatedAtom = Atom(xmlString = xml)
176           
177            self.assertNotEqual(atom, updatedAtom)
178           
179            # there should be one extra link on the new atom
180            self.assertEqual(len(atom.relatedLinks) + 1, len(updatedAtom.relatedLinks))
181        except Exception, e:
182            self.fail("An exception shouldn't have been thrown")
183        finally:
184            self.utils.deleteDoc(atom.getFullPath())
185
186
187    def testAddGranuleToDataEntityRecords(self):
188        # firstly create a DE to use
189        atom = Atom(VTD.DE_TERM)
190        atom.fromString(tc.xmlString)
191        atom.atomTypeID = VTD.DE_TERM
192        atom = self.utils.ac.createAtom(atom)
193
194        # now set up the granulite
195        details = self._initGranule()
196        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
197        self.g._granulite__applyCoreGranuliteDetails(details)
198       
199        # and now update DE
200        self.g._dataEntityIDs = [atom.datasetID]
201        self.g._granulite__addGranuleToDataEntityRecords()
202        atomPath = atom.getDefaultCollectionPath() + atom.atomName
203        self.assertEqual([], self.utils.ac.checkAtomSchemaCompliance(atomPath))
204
205        xml = self.utils.ac.getNDGDoc(atom.ME.providerID, no.ATOM_DOC_TYPE,
206                                      tc.datasetID, 
207                                      targetCollection = dc.ATOM_COLLECTION_PATH)
208        updatedAtom = Atom(xmlString = xml)
209        self.assertNotEqual(atom, updatedAtom)
210       
211        # there should be one extra link on the new atom
212        self.assertEqual(len(atom.relatedLinks) + 1, len(updatedAtom.relatedLinks))
213        self.utils.deleteDoc(atomPath)
214       
215       
216if __name__ == '__main__':
217
218    unittest.main()
Note: See TracBrowser for help on using the repository browser.