source: ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testxmlrpcclient.py @ 5219

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testxmlrpcclient.py@5219
Revision 5219, 10.6 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
2import unittest,os, xmlrpclib
3from ndg.common.src.clients.xmldb.eXist.xmlrpcclient import *
4import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
5import ndg.common.unittests.testconstants as tc
6from ndg.common.unittests.testutils import testUtils as tu
7from ndg.common.src.lib.ndgresources import ndgResources
8import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
9from ndg.common.src.lib.utilities import loadConfigDetails
10
11class TestCase(unittest.TestCase): 
12
13
14    def setUp(self):
15        '''
16        set up data used in the tests.
17        '''
18        userid, password, host = loadConfigDetails(tc.EXIST_DBCONFIG_FILE, tc.EXIST_DB)
19        self.client = XMLRPCClient(host, userid, password)
20
21        self.createdAtoms = []    # array to store paths to atoms created - in order to then delete them
22        self.utils = tu(tc.EXIST_DBCONFIG_FILE)
23
24    def testDescribeCollection(self):
25        info = self.client.describeCollection(dc.ATOM_COLLECTION_PATH)
26        self.assertEqual(dc.ATOM_COLLECTION_PATH, info['name'] + '/')
27        self.assertTrue(len(info['collections']) > 0)
28        self.assertTrue(info['owner'])
29        self.assertTrue(info['created'])
30        self.assertTrue(info['group'])
31        self.assertTrue(info['permissions'])
32       
33    def testRunQuery(self):
34        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
35        self.createdAtoms.append(createdAtom.getFullPath())
36        res = ndgResources()
37        xquery = res.createXQuery('atomList', dc.ATOM_COLLECTION_PATH, "", "")
38        result = self.client.runQuery(xquery)[0]
39        self.assertTrue(len(result) > 0)
40        self.assertTrue(result.find('fileName') > 0)
41
42    def testInvalidRunQuery(self):
43        res = ndgResources()
44        result = self.client.runQuery(tc.INVALID_XQUERY)
45        self.assertEqual([], result)
46
47    def testEmptyExecuteQuery(self):
48        self.assertRaises(xmlrpclib.Fault, self.client._XMLRPCClient__executeQuery, "")
49
50    def testValidExecuteQuery(self):
51        # NB, this test assumes there is at least one atom available in the exist db
52        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
53        self.createdAtoms.append(createdAtom.getFullPath())
54        res = ndgResources()
55        xquery = res.createXQuery('atomList', dc.ATOM_COLLECTION_PATH, "", "")
56        result = self.client._XMLRPCClient__executeQuery(xquery)
57        self.assertTrue(result[1]['hits'] > 0)
58
59
60    def testValidExecuteQueryMultipleResults(self):
61        # NB, this test assumes there is at least one atom available in the exist db
62        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
63        self.createdAtoms.append(createdAtom.getFullPath())
64
65        res = ndgResources()
66        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
67        result = self.client._XMLRPCClient__executeQuery(xquery)
68        noHits = result[1]['hits']
69        self.assertTrue(noHits > 0)
70       
71        atom2 = self.utils.createAtom(tc.xmlString)
72        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
73        atom2.setDatasetID(tc.datasetID)
74        path = atom2.getFullPath()
75        self.client.createCollection(atom2.getDefaultCollectionPath())
76        self.client.createDoc(atom2.toPrettyXML(), path)
77        self.createdAtoms.append(path)
78       
79        result = self.client._XMLRPCClient__executeQuery(xquery)
80        self.assertEquals(noHits + 1, result[1]['hits'])
81
82
83    def testValidRunQueryMultipleResults(self):
84        # NB, this test assumes there is at least one atom available in the exist db
85        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
86        self.createdAtoms.append(createdAtom.getFullPath())
87
88        res = ndgResources()
89        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
90        docs = self.client.runQuery(xquery)
91        self.assertEqual(1, len(docs))
92        self.assertTrue(docs[0])
93        # NB, not sure why but sometimes a newline character gets introduced in
94        # results - remove this, just in case
95        docs[0] = docs[0].strip()
96       
97        atom2 = self.utils.createAtom(tc.xmlString)
98        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
99        atom2.setDatasetID(tc.datasetID)
100        path = atom2.getFullPath()
101        self.client.createCollection(atom2.getDefaultCollectionPath())
102        self.client.createDoc(atom2.toPrettyXML(), path)
103        self.createdAtoms.append(path)
104       
105        docs2 = self.client.runQuery(xquery)
106        self.assertEqual(2, len(docs2))
107        self.assertTrue(docs2[0])
108        self.assertTrue(docs2[1])
109        docs2[0] = docs2[0].strip()
110        docs2[1] = docs2[1].strip()
111        # NB, cannot guarantee order of results
112        self.assertTrue(docs[0] == docs2[0] or docs[0] == docs2[1])
113
114
115    def testValidRunQueryMultipleResultsTwoResultsKeyword(self):
116        # NB, this test assumes there is at least one atom available in the exist db
117        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
118        self.createdAtoms.append(createdAtom.getFullPath())
119
120        res = ndgResources()
121        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
122        docs = self.client.runQuery(xquery)
123        self.assertEqual(1, len(docs))
124        self.assertTrue(docs[0])
125        docs[0] = docs[0].strip()
126       
127        atom2 = self.utils.createAtom(tc.xmlString)
128        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
129        atom2.setDatasetID(tc.datasetID)
130        path = atom2.getFullPath()
131        self.client.createCollection(atom2.getDefaultCollectionPath())
132        self.client.createDoc(atom2.toPrettyXML(), path)
133        self.createdAtoms.append(path)
134       
135        docs2 = self.client.runQuery(xquery)
136        self.assertEqual(2, len(docs2))
137        self.assertTrue(docs2[0])
138        self.assertTrue(docs2[1])
139        docs2[0] = docs2[0].strip()
140        docs2[1] = docs2[1].strip()
141        # NB, cannot guarantee order of results
142        self.assertTrue(docs[0] == docs2[0] or docs[0] == docs2[1])
143
144        docs3 = self.client.runQuery(xquery, noResults = 2)
145        self.assertEqual(2, len(docs3))
146        self.assertTrue(docs3[0])
147        self.assertTrue(docs3[1])
148        # NB, cannot guarantee order of results
149        self.assertTrue(docs[0] == docs3[0] or docs[0] == docs3[1])
150
151
152    def testValidRunQueryMultipleZeroResultsKeywordChunkedOff(self):
153        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
154        self.createdAtoms.append(createdAtom.getFullPath())
155
156        res = ndgResources()
157        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
158        docs = self.client.runQuery(xquery, noResults = 0, useChunked = False)
159        self.assertEqual(0, len(docs))
160
161
162    def testValidRunQueryMultipleZeroResultsKeywordChunkedOn(self):
163        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
164        self.createdAtoms.append(createdAtom.getFullPath())
165
166        res = ndgResources()
167        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
168        docs = self.client.runQuery(xquery, noResults = 0)
169        self.assertEqual(1, len(docs))
170
171
172    def testValidRunQueryMultipleResultsOneResultKeyword(self):
173        # NB, this test assumes there is at least one atom available in the exist db
174        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
175        self.createdAtoms.append(createdAtom.getFullPath())
176
177        res = ndgResources()
178        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
179       
180        atom2 = self.utils.createAtom(tc.xmlString)
181        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
182        atom2.setDatasetID(tc.datasetID)
183        path = atom2.getFullPath()
184        self.client.createCollection(atom2.getDefaultCollectionPath())
185        self.client.createDoc(atom2.toPrettyXML(), path)
186        self.createdAtoms.append(path)
187       
188        docs2 = self.client.runQuery(xquery, noResults = 1)
189        self.assertEqual(1, len(docs2))
190        self.assertTrue(docs2[0])
191
192
193
194    def testInvalidExecuteQuery(self):
195        # NB, this test assumes there is at least one atom available in the exist db
196        res = ndgResources()
197        result = self.client._XMLRPCClient__executeQuery(tc.INVALID_XQUERY)
198        self.assertEquals(0, result[1]['hits'])
199
200
201    def testRetrieve(self):
202        res = ndgResources()
203        xquery = res.createXQuery('atomList', dc.ATOM_COLLECTION_PATH, "", "")
204        result = self.client._XMLRPCClient__executeQuery(xquery)
205        doc = self.client._XMLRPCClient__retrieve(result[0], 0)
206        self.assertTrue(doc)
207
208
209    def testCreateCollection(self):
210        self.client.removeCollection(tc.TEST_COLLECTION_PATH)
211        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
212        self.assertTrue(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
213
214
215    def testDuplicateCreateCollection(self):
216        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
217        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
218        self.assertTrue(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
219       
220
221    def testRemoveCollection(self):
222        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
223        self.assertTrue(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
224
225
226    def testInvalidRemoveCollection(self):
227        self.assertFalse(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
228   
229    def testCreateDoc(self):
230        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
231        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
232   
233    def testCreateDocWithoutOverwrite(self):
234        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
235        self.assertRaises(xmlrpclib.Fault, self.client.createDoc, tc.xmlString, tc.TEST_FILE_NAME)
236        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
237   
238    def testCreateDocWithOverwrite(self):
239        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
240        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME, overwrite=1))
241        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
242
243    def testRemoveDoc(self):
244        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
245        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
246
247    def testInvalidRemoveDoc(self):
248        self.assertRaises(xmlrpclib.Fault, self.client.removeDoc, tc.TEST_FILE_NAME)
249
250       
251    def tearDown(self):
252        for path  in self.createdAtoms:
253            # delete the test file, in case it has been used in a test
254            self.utils.deleteDoc(path)
255
256           
257if __name__=="__main__":
258    unittest.main()
Note: See TracBrowser for help on using the repository browser.