CoCalc -- test

GitHub Repository: singlestore-labs/singlestoredb-python
Path: blob/main/singlestoredb/tests/test_http.py
⁴⁶⁹ views
1
#!/usr/bin/env python
2
# type: ignore
3
"""SingleStoreDB HTTP connection testing."""
4
import base64
5
import os
6
import unittest
7

8
import singlestoredb.connection as sc
9
from singlestoredb import config
10
from singlestoredb import http
11
from singlestoredb.tests import utils
12
# import traceback
13

14

15
class TestHTTP(unittest.TestCase):
16

17
    dbname: str = ''
18
    dbexisted: bool = False
19

20
    @classmethod
21
    def setUpClass(cls):
22
        sql_file = os.path.join(os.path.dirname(__file__), 'test.sql')
23
        cls.dbname, cls.dbexisted = utils.load_sql(sql_file)
24

25
    @classmethod
26
    def tearDownClass(cls):
27
        if not cls.dbexisted:
28
            utils.drop_database(cls.dbname)
29

30
    def setUp(self):
31
        self.conn = self._connect()
32
        self.cur = self.conn.cursor()
33
        if self.params['driver'] not in ['http', 'https']:
34
            self.skipTest('Tests must be run using HTTP connection')
35
        self.driver = self.params['driver'] or 'http'
36

37
    def _connect(self):
38
        params = sc.build_params(host=config.get_option('host'))
39
        self.params = {
40
            k: v for k, v in dict(
41
                host=params.get('host'),
42
                port=params.get('port'),
43
                user=params.get('user'),
44
                password=params.get('password'),
45
                driver=params.get('driver'),
46
            ).items() if v is not None
47
        }
48
        return http.connect(database=type(self).dbname, **self.params)
49

50
    def tearDown(self):
51
        try:
52
            if self.cur is not None:
53
                self.cur.close()
54
        except Exception:
55
            # traceback.print_exc()
56
            pass
57

58
        try:
59
            if self.conn is not None:
60
                self.conn.close()
61
        except Exception:
62
            # traceback.print_exc()
63
            pass
64

65
    def test_get_exc_type(self):
66
        exc = http.connection.get_exc_type(0)
67
        assert exc is http.InterfaceError, exc
68

69
        exc = http.connection.get_exc_type(2012)
70
        assert exc is http.InterfaceError, exc
71

72
        exc = http.connection.get_exc_type(1230)
73
        assert exc is http.DataError, exc
74

75
        exc = http.connection.get_exc_type(1110)
76
        assert exc is http.ProgrammingError, exc
77

78
        exc = http.connection.get_exc_type(1452)
79
        assert exc is http.IntegrityError, exc
80

81
        exc = http.connection.get_exc_type(9999)
82
        assert exc is http.OperationalError, exc
83

84
        exc = http.connection.get_exc_type(222)
85
        assert exc is http.InternalError, exc
86

87
    def test_identity(self):
88
        out = http.connection.identity(1)
89
        assert out == 1, out
90

91
        out = http.connection.identity('hi')
92
        assert out == 'hi', out
93

94
    def test_b64decode_converter(self):
95
        data = base64.b64encode(b'hi there')
96
        assert type(data) is bytes, type(data)
97

98
        out = http.connection.b64decode_converter(http.connection.identity, None)
99
        assert out is None, out
100

101
        out = http.connection.b64decode_converter(http.connection.identity, data)
102
        assert out == b'hi there', out
103

104
        out = http.connection.b64decode_converter(
105
            http.connection.identity,
106
            str(data, 'utf8'),
107
        )
108
        assert out == b'hi there', out
109

110
    def test_executemany(self):
111
        self.cur.executemany(
112
            'select * from data where id < %(id)s',
113
            [dict(id='d'), dict(id='e')],
114
        )
115

116
        assert self.cur.rownumber == 0, self.cur.rownumber
117

118
        # First set
119
        out = self.cur.fetchall()
120

121
        assert self.cur.rownumber == 3, self.cur.rownumber
122

123
        desc = self.cur.description
124
        rowcount = self.cur.rowcount
125
        lastrowid = self.cur.lastrowid
126

127
        assert sorted(out) == sorted([
128
            ('a', 'antelopes', 2),
129
            ('b', 'bears', 2),
130
            ('c', 'cats', 5),
131
        ]), out
132

133
        assert rowcount == 7, rowcount
134
        assert lastrowid is None, lastrowid
135
        assert len(desc) == 3, desc
136
        assert desc[0][0] == 'id', desc[0][0]
137
        assert desc[0][1] in [253, 15], desc[0][1]
138
        assert desc[1][0] == 'name', desc[1][0]
139
        assert desc[1][1] in [253, 15], desc[1][1]
140
        assert desc[2][0] == 'value', desc[2][0]
141
        assert desc[2][1] == 8, desc[2][1]
142

143
        # Second set
144
        self.cur.nextset()
145

146
        out = self.cur.fetchall()
147

148
        desc = self.cur.description
149
        rowcount = self.cur.rowcount
150
        lastrowid = self.cur.lastrowid
151

152
        assert sorted(out) == sorted([
153
            ('a', 'antelopes', 2),
154
            ('b', 'bears', 2),
155
            ('c', 'cats', 5),
156
            ('d', 'dogs', 4),
157
        ]), out
158

159
        assert rowcount == 4, rowcount
160
        assert lastrowid is None, lastrowid
161
        assert len(desc) == 3, desc
162
        assert desc[0][0] == 'id', desc[0][0]
163
        assert desc[0][1] in [253, 15], desc[0][1]
164
        assert desc[1][0] == 'name', desc[1][0]
165
        assert desc[1][1] in [253, 15], desc[1][1]
166
        assert desc[2][0] == 'value', desc[2][0]
167
        assert desc[2][1] == 8, desc[2][1]
168

169
        out = self.cur.nextset()
170
        assert out is None, out
171

172
    def test_executemany_no_args(self):
173
        self.cur.executemany('select * from data where id < "d"')
174

175
        # First set
176
        out = self.cur.fetchall()
177

178
        desc = self.cur.description
179
        rowcount = self.cur.rowcount
180
        lastrowid = self.cur.lastrowid
181

182
        assert sorted(out) == sorted([
183
            ('a', 'antelopes', 2),
184
            ('b', 'bears', 2),
185
            ('c', 'cats', 5),
186
        ]), out
187

188
        assert rowcount == 3, rowcount
189
        assert lastrowid is None, lastrowid
190
        assert len(desc) == 3, desc
191
        assert desc[0][0] == 'id', desc[0][0]
192
        assert desc[0][1] in [253, 15], desc[0][1]
193
        assert desc[1][0] == 'name', desc[1][0]
194
        assert desc[1][1] in [253, 15], desc[1][1]
195
        assert desc[2][0] == 'value', desc[2][0]
196
        assert desc[2][1] == 8, desc[2][1]
197

198
        out = self.cur.nextset()
199
        assert out is None, out
200

201
    def test_is_connected(self):
202
        assert self.cur.is_connected() is True
203
        self.cur.close()
204
        assert self.cur.is_connected() is False
205

206
    def test_close(self):
207
        self.cur.close()
208
        assert self.cur.is_connected() is False
209

210
        with self.assertRaises(http.ProgrammingError):
211
            self.cur.execute('select 1')
212

213
        with self.assertRaises(http.ProgrammingError):
214
            self.cur.executemany('select 1')
215

216
        with self.assertRaises(http.ProgrammingError):
217
            self.cur.callproc('get_animal', ['cats'])
218

219
#   def test_callproc(self):
220
#       with self.assertRaises(NotImplementedError):
221
#           self.cur.callproc('get_animal', ['cats'])
222

223
    def test_iter(self):
224
        self.cur.execute('select * from data')
225

226
        out = list(self.cur)
227

228
        assert sorted(out) == sorted([
229
            ('a', 'antelopes', 2),
230
            ('b', 'bears', 2),
231
            ('c', 'cats', 5),
232
            ('d', 'dogs', 4),
233
            ('e', 'elephants', 0),
234
        ]), out
235

236
    def test_next(self):
237
        self.cur.execute('select * from data')
238

239
        out = [next(self.cur) for i in range(5)]
240

241
        assert sorted(out) == sorted([
242
            ('a', 'antelopes', 2),
243
            ('b', 'bears', 2),
244
            ('c', 'cats', 5),
245
            ('d', 'dogs', 4),
246
            ('e', 'elephants', 0),
247
        ]), out
248

249
        with self.assertRaises(StopIteration):
250
            next(self.cur)
251

252
    def test_context_manager(self):
253
        with self._connect() as conn:
254
            with conn.cursor() as cur:
255
                assert conn.is_connected()
256
                assert cur.is_connected()
257
        assert not conn.is_connected()
258
        assert not cur.is_connected()
259

260
    def test_commit(self):
261
        self.conn.autocommit(True)
262
        assert self.conn.commit() is None
263
        self.conn.autocommit(False)
264
        with self.assertRaises(http.NotSupportedError):
265
            self.conn.commit()
266

267
    def test_rollback(self):
268
        self.conn.autocommit(True)
269
        assert self.conn.rollback() is None
270
        self.conn.autocommit(False)
271
        with self.assertRaises(http.NotSupportedError):
272
            self.conn.rollback()
273

274
    def test_http_error(self):
275
        # Break content type
276
        self.conn._sess.headers.update({
277
            'Content-Type': 'GaRbAge',
278
        })
279
        with self.assertRaises(http.InternalError) as cm:
280
            self.cur.execute('select 1')
281
        exc = cm.exception
282
        assert exc.errno == 415, exc.errno
283
        assert 'Content-Type' in exc.msg, exc.msg
284

285

286
if __name__ == '__main__':
287
    import nose2
288
    nose2.main()
289

290
Product

Resources

Company
