Codebase list python-cx-oracle / master test / SodaDatabase.py
master

Tree @master (Download .tar.gz)

SodaDatabase.py @master — raw · history · blame 

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
#------------------------------------------------------------------------------
# Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
#------------------------------------------------------------------------------

"""Module for testing Simple Oracle Document Access (SODA) Database"""

import TestEnv

import cx_Oracle
import json

class TestCase(TestEnv.BaseTestCase):

    def __dropExistingCollections(self, sodaDatabase):
        for name in sodaDatabase.getCollectionNames():
            sodaDatabase.openCollection(name).drop()

    def __verifyDocument(self, doc, rawContent, strContent=None, content=None,
            key=None, mediaType='application/json'):
        self.assertEqual(doc.getContentAsBytes(), rawContent)
        if strContent is not None:
            self.assertEqual(doc.getContentAsString(), strContent)
        if content is not None:
            self.assertEqual(doc.getContent(), content)
        self.assertEqual(doc.key, key)
        self.assertEqual(doc.mediaType, mediaType)

    def testCreateDocumentWithJson(self):
        "test creating documents with JSON data"
        sodaDatabase = self.connection.getSodaDatabase()
        val = {"testKey1" : "testValue1", "testKey2" : "testValue2" }
        strVal = json.dumps(val)
        bytesVal = strVal.encode("UTF-8")
        key = "MyKey"
        mediaType = "text/plain"
        doc = sodaDatabase.createDocument(val)
        self.__verifyDocument(doc, bytesVal, strVal, val)
        doc = sodaDatabase.createDocument(strVal, key)
        self.__verifyDocument(doc, bytesVal, strVal, val, key)
        doc = sodaDatabase.createDocument(bytesVal, key, mediaType)
        self.__verifyDocument(doc, bytesVal, strVal, val, key, mediaType)

    def testCreateDocumentWithRaw(self):
        "test creating documents with raw data"
        sodaDatabase = self.connection.getSodaDatabase()
        val = b"<html/>"
        key = "MyRawKey"
        mediaType = "text/html"
        doc = sodaDatabase.createDocument(val)
        self.__verifyDocument(doc, val)
        doc = sodaDatabase.createDocument(val, key)
        self.__verifyDocument(doc, val, key=key)
        doc = sodaDatabase.createDocument(val, key, mediaType)
        self.__verifyDocument(doc, val, key=key, mediaType=mediaType)

    def testGetCollectionNames(self):
        "test getting collection names from the database"
        sodaDatabase = self.connection.getSodaDatabase()
        self.__dropExistingCollections(sodaDatabase)
        self.assertEqual(sodaDatabase.getCollectionNames(), [])
        names = ["zCol", "dCol", "sCol", "aCol", "gCol"]
        sortedNames = list(sorted(names))
        for name in names:
            sodaDatabase.createCollection(name)
        self.assertEqual(sodaDatabase.getCollectionNames(), sortedNames)
        self.assertEqual(sodaDatabase.getCollectionNames(limit=2),
                sortedNames[:2])
        self.assertEqual(sodaDatabase.getCollectionNames("a"), sortedNames)
        self.assertEqual(sodaDatabase.getCollectionNames("C"), sortedNames)
        self.assertEqual(sodaDatabase.getCollectionNames("b", limit=3),
                sortedNames[1:4])
        self.assertEqual(sodaDatabase.getCollectionNames("z"),
                sortedNames[-1:])

    def testOpenCollection(self):
        "test opening a collection"
        sodaDatabase = self.connection.getSodaDatabase()
        self.__dropExistingCollections(sodaDatabase)
        coll = sodaDatabase.openCollection("CollectionThatDoesNotExist")
        self.assertEqual(coll, None)
        createdColl = sodaDatabase.createCollection("cxoTestOpenCollection")
        coll = sodaDatabase.openCollection(createdColl.name)
        self.assertEqual(coll.name, createdColl.name)
        coll.drop()

    def testRepr(self):
        "test SodaDatabase representation"
        con1 = self.connection
        con2 = TestEnv.GetConnection()
        sodaDatabase1 = con1.getSodaDatabase()
        sodaDatabase2 = con1.getSodaDatabase()
        sodaDatabase3 = con2.getSodaDatabase()
        self.assertEqual(str(sodaDatabase1), str(sodaDatabase2))
        self.assertEqual(str(sodaDatabase2), str(sodaDatabase3))

    def testNegative(self):
        "test negative cases for SODA database methods"
        sodaDatabase = self.connection.getSodaDatabase()
        self.assertRaises(TypeError, sodaDatabase.createCollection)
        self.assertRaises(TypeError, sodaDatabase.createCollection, 1)
        self.assertRaises(cx_Oracle.DatabaseError,
                sodaDatabase.createCollection, None)
        self.assertRaises(TypeError, sodaDatabase.getCollectionNames, 1)

if __name__ == "__main__":
    TestEnv.RunTestCases()