source: ndgCommon/trunk/ndg/common/unittests/models/testatom.py @ 4810

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/models/testatom.py@4810
Revision 4810, 10.8 KB checked in by cbyrom, 12 years ago (diff)

Create new class (and test suite) to do all the eXist DB setting up - and move this code from the
eXistdbclient to tidy things up.
Allow feed publishing to be done synch/asynch.
Improve loading of non-python resources - adding code to store the various schemata and indexes
required for setting up eXist
Improve parsing of atom URLs to allow browser host to change
Fix moles data in testconstants
Add tests for looking up associated atom data
Add indexing configuration files - to optimise atom processing in eXist + add code in initialiser class to set these up on eXist.
Add all the required schemata required for atom schema validation in eXist + add code in initialiser class to set these up on eXist;
this allows eXist to be ran without need of access beyond the firewall.

Line 
1'''
2A test class for the GranuleAtom module.
3
4C Byrom Jun 2008
5Tessella
6'''
7import unittest, copy
8from ndg.common.src.models.Atom import Atom, Link
9from ndg.common.src.models.vocabtermdata import VocabTermData as VTD
10from ndg.common.src.lib.utilities import getTripleData
11import ndg.common.unittests.testconstants as tc
12from ndg.common.src.models import AtomState
13from ndg.common.unittests.testutils import testUtils as tu
14from ndg.common.src.dal.DocumentRetrieve import DocumentRetrieve
15
16class testatom(unittest.TestCase):
17    VTD = VTD()
18   
19    def setUp(self):
20        '''
21        set up data used in the tests.
22        '''
23        print "Setting up..."
24        self.atom = Atom(VTD.GRANULE_TERM, vocabTermData = self.VTD)
25        self.link = Link()
26        self.link.title = 'fred'
27        self.link.href = 'www.fred.com'
28        self.link.rel = 'related'
29        self.link2 = Link()
30        self.link2.title = 'pete'
31        self.link2.href = 'www.pete.com'
32        self.link2.rel = 'self'
33        self.createdAtoms = []    # array to store paths to atoms created - in order to then delete them
34        self.utils = tu(tc.DBCONFIG_FILE)
35
36               
37    def testAddValidParameters(self):
38        self.atom.addParameters(tc.validParams)
39        self.assertEqual(self.atom.parameters[0].term, getTripleData(tc.validParams[0])[2])
40        self.assertEqual(self.atom.parameters[1].term, getTripleData(tc.validParams[1])[2])
41        self.assertEqual(self.atom.parameters[2].term, getTripleData(tc.validParams[2])[2])
42       
43       
44    def testAddMessyParameters(self):
45        self.atom.addParameters(tc.messyParams)
46        self.assertEqual(self.atom.parameters[0].term, 'OZONE')
47        self.assertEqual(self.atom.parameters[1].term, 'ERROR')
48        self.assertEqual(self.atom.parameters[2].term, 'HYDROGEN &')
49        self.assertEqual(self.atom.parameters[2].scheme, 'http://vocab.ndg.nerc.ac.uk/blah')
50        self.assertEqual(self.atom.parameters[2].label.lower(), 'Hydrogen Peroxide < ppbv'.lower())
51       
52    def testFromString(self):
53        self.atom.fromString(tc.xmlString)
54       
55        self.assertNotEqual(self.atom.author, None)
56        self.assertEqual(tc.validAuthor, self.atom.author.name)
57        self.assertEqual(3, len(self.atom.parameters))
58       
59        for param in self.atom.parameters:
60            foundIt = False
61            for validParam in tc.validParams:
62                if (validParam.startswith(param.term)):
63                    foundIt = True
64                    break
65           
66            self.assertEqual(True, foundIt, "Cound not find parameter (%s) in valid parameter list" %param)
67       
68        # currently only store max/min boundaries
69        #for spatial in self.atom.spatialData:
70        #    if not spatial in self.validSpatialData:
71        #        self.fail("Cound not find spatial data (%s) in valid coverage list" %spatial)
72        #
73        #for temporal in self.atom.temporalData:
74        #    if not temporal in self.validTemporalData:
75        #        self.fail("Cound not find temporal data (%s) in valid coverage list" %temporal)
76       
77        #self.assertEqual(len(self.atom.relatedLinks), 2)
78        for link in self.atom.relatedLinks:
79            foundIt = False
80            for validLink in tc.validLinks:
81                if (validLink.startswith(link.href)):
82                    foundIt = True
83                    break
84           
85            self.assertEqual(True, foundIt, "Cound not find link (%s) in valid related link list" %link.href)
86
87        self.assertEqual(tc.summaryText, self.atom.summary[0])
88        self.assertEqual(tc.csmlFile, self.atom.csmlFile)
89        self.assertEqual(None, self.atom.cdmlFile)
90        self.assertEqual(tc.title, self.atom.title)
91        self.assertEqual(tc.logo, self.atom.getLogos()[0].href)
92        self.assertEqual(tc.datasetID, self.atom.datasetID)
93        self.assertEqual(tc.minX, self.atom.minX)
94        self.assertEqual(tc.minY, self.atom.minY)
95        self.assertEqual(tc.maxX, self.atom.maxX)
96        self.assertEqual(tc.maxY, self.atom.maxY)
97        self.assertEqual(tc.T1, self.atom.t1)
98        self.assertEqual(tc.T2, self.atom.t2)
99       
100        # date when the granule was first ingested
101        if not self.atom.publishedDate:
102            self.fail("No published date found") 
103
104        self.assertEqual(tc.name, self.atom.atomName)
105       
106
107    def testImportExport(self):
108        self.atom.fromString(tc.xmlString)
109        xml = self.atom.toPrettyXML()
110        # NB, this is a naive test as ordering of sibling elements changes as does addition of new line characters
111        #self.assertEqual(xml, self.xmlString)
112
113    def testCreateWithInputDict(self):
114        inputs = {
115                  'atomTypeID': VTD.ACTIVITY_TERM,
116                  'title': tc.title
117                  }
118        a = Atom(**inputs)
119        self.assertEqual(tc.title, a.title)
120        self.assertEqual(VTD.ACTIVITY_TERM, a.atomTypeID)
121        self.assertEqual(VTD.TERM_DATA[VTD.ACTIVITY_TERM].title, a.atomTypeName)
122
123    def testaddUniqueRelatedLinks(self):
124        self.atom.fromString(tc.xmlString)
125        linkNo = len(self.atom.relatedLinks)
126        self.atom.addUniqueRelatedLinks(self.link)
127        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
128        self.assertTrue(self.link in self.atom.relatedLinks)
129
130    def testAddMultipeDifferentRelatedLinks(self):
131        self.atom.fromString(tc.xmlString)
132        linkNo = len(self.atom.relatedLinks)
133        self.atom.addUniqueRelatedLinks(self.link)
134        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
135        link2 = copy.deepcopy(self.link)
136        link2.title = "blabby"
137        self.atom.addUniqueRelatedLinks(link2)
138        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
139        self.atom.addUniqueRelatedLinks(link2)
140        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
141
142    def testAddMultipeDifferentRelatedLinksSimultaneously(self):
143        self.atom.fromString(tc.xmlString)
144        linkNo = len(self.atom.relatedLinks)
145        link2 = copy.deepcopy(self.link)
146        link2.title = "blabby"
147        links = [self.link, link2]
148        self.atom.addUniqueRelatedLinks(links)
149        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
150        self.assertTrue(self.link in self.atom.relatedLinks)
151        self.assertTrue(link2 in self.atom.relatedLinks)
152
153        self.atom.addUniqueRelatedLinks(link2)
154        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
155       
156
157    def testAddMultipleUniqueRelatedLinks(self):
158        self.atom.fromString(tc.xmlString)
159        linkNo = len(self.atom.relatedLinks)
160        self.atom.addUniqueRelatedLinks(self.link)
161        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
162        self.atom.addUniqueRelatedLinks(self.link)
163        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
164        self.atom.addUniqueRelatedLinks(self.link)
165        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
166
167    def testRemoveRelatedLinks(self):
168        self.atom.fromString(tc.xmlString)
169        linkNo = len(self.atom.relatedLinks)
170        self.atom.addUniqueRelatedLinks(self.link)
171        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
172        self.assertTrue(self.link in self.atom.relatedLinks)
173        self.atom.removeRelatedLinks(self.link)
174        self.assertEqual(linkNo, len(self.atom.relatedLinks))
175        self.assertFalse(self.link in self.atom.relatedLinks)
176
177    def testRemoveMultipleRelatedLinks(self):
178        self.atom.fromString(tc.xmlString)
179        linkNo = len(self.atom.relatedLinks)
180        link2 = copy.deepcopy(self.link)
181        link2.title = "blabby"
182        links = [self.link, link2]
183
184        self.atom.addUniqueRelatedLinks(links)
185        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
186        self.assertTrue(self.link in self.atom.relatedLinks)
187        self.assertTrue(link2 in self.atom.relatedLinks)
188
189        self.atom.removeRelatedLinks(links)
190        self.assertEqual(linkNo, len(self.atom.relatedLinks))
191        self.assertFalse(self.link in self.atom.relatedLinks)
192        self.assertFalse(link2 in self.atom.relatedLinks)
193
194    def testAddUniqueLinks(self):
195        data = []
196        links = [self.link, self.link2, self.link]
197        self.atom.addUniqueLinks(data, links)
198        self.assertEqual(2, len(data))
199
200    def testAddUniqueLinksWithInitialLinksDefined(self):
201        data = [self.link2]
202        links = [self.link, self.link2, self.link]
203        self.atom.addUniqueLinks(data, links)
204        self.assertEqual(2, len(data))
205
206    def testIsPublished(self):
207        self.assertFalse(self.atom.isPublished())
208        self.atom.state = AtomState.PUBLISHED_STATE
209        self.assertTrue(self.atom.isPublished())
210       
211    def testStateKeyword(self):
212        self.atom = Atom(VTD.GRANULE_TERM, vocabTermData = self.VTD, 
213                         state = AtomState.OLD_STATE)
214        self.assertEqual(AtomState.OLD_STATE, self.atom.state)
215        self.assertFalse(self.atom.isPublished())
216
217    def testLookupAssociatedDEData(self):
218        atom = self.utils.createAtomInEXist(tc.invalidXmlString)
219        self.createdAtoms.append(atom.getFullPath())
220        dr = DocumentRetrieve(tc.VALID_REPOSITORY,
221                              pwfile = tc.DBCONFIG_FILE)
222
223        atom.lookupAssociatedData(VTD.DE_TERM, dr)
224        self.utils.dbc.deleteAtomInExist(atom)
225
226    def testLookupAssociatedDeploymentData(self):
227        atom = self.utils.createAtomInEXist(tc.invalidXmlString)
228        self.createdAtoms.append(atom.getFullPath())
229        dr = DocumentRetrieve(tc.VALID_REPOSITORY,
230                              pwfile = tc.DBCONFIG_FILE)
231
232        atom.lookupAssociatedData(VTD.DEPLOYMENT_TERM, dr)
233        self.utils.dbc.deleteAtomInExist(atom)
234
235    def testLookupAssociatedDeploymentDataWithLookupIndirectReferences(self):
236        atom = self.utils.createAtomInEXist(tc.invalidXmlString)
237        self.createdAtoms.append(atom.getFullPath())
238        dr = DocumentRetrieve(tc.VALID_REPOSITORY,
239                              pwfile = tc.DBCONFIG_FILE)
240
241        atom.lookupAssociatedData(VTD.DEPLOYMENT_TERM, dr,
242                                  lookupIndirectReferences=True)
243        self.utils.dbc.deleteAtomInExist(atom)
244
245    def testLookupAssociatedDEDataWithLookupIndirectReferences(self):
246        atom = self.utils.createAtomInEXist(tc.invalidXmlString)
247        self.createdAtoms.append(atom.getFullPath())
248        dr = DocumentRetrieve(tc.VALID_REPOSITORY,
249                              pwfile = tc.DBCONFIG_FILE)
250
251        atom.lookupAssociatedData(VTD.DE_TERM, dr,
252                                  lookupIndirectReferences=True)
253        self.utils.dbc.deleteAtomInExist(atom)
254
255       
256
257if __name__ == '__main__':
258
259    unittest.main()
Note: See TracBrowser for help on using the repository browser.