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

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testxmlrpcclient.py@4942
Revision 4942, 9.9 KB checked in by cbyrom, 11 years ago (diff)

Replace old eXist connector tests with tests based on the new clients
+ extend and fix this wherever necessary.

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