mirror of
https://github.com/hanzo-ds/driver.git
synced 2026-10-30 20:32:40 +00:00
zeekay
eb74c996a6
Imports read `datastore_driver`, distribution `datastore-driver`, so nothing has to import another product's name to talk to Hanzo Datastore. Cut from upstream 0.2.9 — the version this stack already ran — rather than from upstream HEAD, so it is a PURE RENAME with no behaviour change. Forking from HEAD would have dragged versions along with the name (connect alone jumped 0.8.17 -> 1.6.0, a major), which is a different and much riskier change wearing a rename's clothes. X-ClickHouse-* headers are untouched: they are wire protocol the server sends and expects, and a different token shape from the module name.
178 lines
5.2 KiB
Python
178 lines
5.2 KiB
Python
from enum import IntEnum
|
|
|
|
from tests.testcase import BaseTestCase
|
|
from datastore_driver import errors
|
|
|
|
|
|
class A(IntEnum):
|
|
hello = -1
|
|
world = 2
|
|
|
|
|
|
class B(IntEnum):
|
|
foo = -300
|
|
bar = 300
|
|
|
|
|
|
class EnumTestCase(BaseTestCase):
|
|
def test_simple(self):
|
|
columns = (
|
|
"a Enum8('hello' = -1, 'world' = 2), "
|
|
"b Enum16('foo' = -300, 'bar' = 300)"
|
|
)
|
|
|
|
data = [(A.hello, B.bar), (A.world, B.foo), (-1, 300), (2, -300)]
|
|
with self.create_table(columns):
|
|
self.client.execute(
|
|
'INSERT INTO test (a, b) VALUES', data
|
|
)
|
|
|
|
query = 'SELECT * FROM test'
|
|
inserted = self.emit_cli(query)
|
|
self.assertEqual(
|
|
inserted, (
|
|
'hello\tbar\n'
|
|
'world\tfoo\n'
|
|
'hello\tbar\n'
|
|
'world\tfoo\n'
|
|
)
|
|
)
|
|
|
|
inserted = self.client.execute(query)
|
|
self.assertEqual(
|
|
inserted, [
|
|
('hello', 'bar'), ('world', 'foo'),
|
|
('hello', 'bar'), ('world', 'foo')
|
|
]
|
|
)
|
|
|
|
def test_enum_by_string(self):
|
|
columns = "a Enum8('hello' = 1, 'world' = 2)"
|
|
data = [('hello', ), ('world', )]
|
|
with self.create_table(columns):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
query = 'SELECT * FROM test'
|
|
inserted = self.emit_cli(query)
|
|
self.assertEqual(
|
|
inserted, (
|
|
'hello\n'
|
|
'world\n'
|
|
)
|
|
)
|
|
|
|
inserted = self.client.execute(query)
|
|
self.assertEqual(inserted, data)
|
|
|
|
def test_errors(self):
|
|
columns = "a Enum8('test' = 1, 'me' = 2)"
|
|
data = [(A.world, )]
|
|
with self.create_table(columns):
|
|
with self.assertRaises(errors.LogicalError):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
columns = "a Enum8('test' = 1, 'me' = 2)"
|
|
data = [(3, )]
|
|
with self.create_table(columns):
|
|
with self.assertRaises(errors.LogicalError):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
def test_quote_in_name(self):
|
|
columns = "a Enum8(' \\' t = ' = -1, 'test' = 2)"
|
|
data = [(-1, ), (" ' t = ", )]
|
|
with self.create_table(columns):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
query = 'SELECT * FROM test'
|
|
inserted = self.emit_cli(query)
|
|
self.assertEqual(
|
|
inserted, (
|
|
" \\' t = \n"
|
|
" \\' t = \n"
|
|
)
|
|
)
|
|
|
|
inserted = self.client.execute(query)
|
|
self.assertEqual(inserted, [(" ' t = ", ), (" ' t = ", )])
|
|
|
|
def test_comma_and_space_in_name(self):
|
|
columns = "a Enum8('one' = 1, 'two_with_comma, ' = 2, 'three' = 3)"
|
|
data = [(2, ), ('two_with_comma, ', )]
|
|
with self.create_table(columns):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
query = 'SELECT * FROM test'
|
|
inserted = self.emit_cli(query)
|
|
|
|
self.assertEqual(
|
|
inserted, (
|
|
'two_with_comma, \n'
|
|
'two_with_comma, \n'
|
|
)
|
|
)
|
|
|
|
inserted = self.client.execute(query)
|
|
self.assertEqual(
|
|
inserted, [('two_with_comma, ',), ('two_with_comma, ',)]
|
|
)
|
|
|
|
def test_nullable(self):
|
|
columns = "a Nullable(Enum8('hello' = -1, 'world' = 2))"
|
|
|
|
data = [(None, ), (A.hello, ), (None, ), (A.world, )]
|
|
with self.create_table(columns):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
query = 'SELECT * FROM test'
|
|
inserted = self.emit_cli(query)
|
|
self.assertEqual(
|
|
inserted, (
|
|
'\\N\nhello\n\\N\nworld\n'
|
|
)
|
|
)
|
|
|
|
inserted = self.client.execute(query)
|
|
self.assertEqual(
|
|
inserted, [
|
|
(None, ), ('hello', ), (None, ), ('world', ),
|
|
]
|
|
)
|
|
|
|
def test_invalid_python_names(self):
|
|
data = [(1, ), (2, ), (3, ), ('', ), ('mro', )]
|
|
with self.create_table("a Enum8('mro' = 1, '' = 2, 'test' = 3)"):
|
|
self.client.execute(
|
|
'INSERT INTO test (a) VALUES', data
|
|
)
|
|
|
|
query = 'SELECT * FROM test'
|
|
inserted = self.emit_cli(query)
|
|
self.assertEqual(
|
|
inserted, (
|
|
'mro\n'
|
|
'\n'
|
|
'test\n'
|
|
'\n'
|
|
'mro\n'
|
|
)
|
|
)
|
|
|
|
inserted = self.client.execute(query)
|
|
self.assertEqual(
|
|
inserted, [
|
|
('mro', ), ('', ), ('test', ), ('', ), ('mro', )
|
|
]
|
|
)
|