Skip to content

Commit

Permalink
Rework tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@DifferentialOrange
DifferentialOrange committed Mar 31, 2022
1 parent ea80949 commit 4ec6748
Show file tree
Hide file tree
Showing 2 changed files with 112 additions and 90 deletions.
4 changes: 3 additions & 1 deletion test/suites/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,13 @@
from .test_mesh import TestSuite_Mesh
from .test_execute import TestSuite_Execute
from .test_dbapi import TestSuite_DBAPI
from .test_encoding import TestSuite_Encoding

test_cases = (TestSuite_Schema_UnicodeConnection,
TestSuite_Schema_BinaryConnection,
TestSuite_Request, TestSuite_Protocol, TestSuite_Reconnect,
TestSuite_Mesh, TestSuite_Execute, TestSuite_DBAPI)
TestSuite_Mesh, TestSuite_Execute, TestSuite_DBAPI,
TestSuite_Encoding)

def load_tests(loader, tests, pattern):
suite = unittest.TestSuite()
Expand Down
198 changes: 109 additions & 89 deletions test/suites/test_encoding.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,94 +10,90 @@
from .lib.tarantool_server import TarantoolServer

class TestSuite_Encoding(unittest.TestCase):
def prepare_dml_test_spaces(self):
self.space_created = self.adm("box.schema.create_space('space_1')")
self.adm("""
box.space['space_1']:create_index('primary', {
type = 'tree',
parts = {1, 'num'},
unique = true})
""".replace('\n', ' '))
self.adm("""
box.space['space_1']:create_index('secondary', {
type = 'tree',
parts = {2, 'num', 3, 'str'},
unique = false})
""".replace('\n', ' '))

self.space_created = self.adm("box.schema.create_space('space_2')")
self.adm("""
box.space['space_2']:create_index('primary', {
type = 'hash',
parts = {1, 'num'},
unique = true})
""".replace('\n', ' '))

def prepare_encoding_test_spaces(self):


def require_modules(self):
self.adm("json = require('json')")
self.adm("fiber = require('fiber')")
self.adm("uuid = require('uuid')")

@classmethod
def setUpClass(self):
print(' DML '.center(70, '='), file=sys.stderr)
print(' ENCODING '.center(70, '='), file=sys.stderr)
print('-' * 70, file=sys.stderr)
self.srv = TarantoolServer()
self.srv.script = 'test/suites/box.lua'
self.srv.start()

self.srv.admin("""
box.schema.user.create('test', { password = 'test' })
box.schema.user.grant('test', 'execute,read,write', 'universe')
box.schema.user.create('test', { password = 'test' })
box.schema.user.grant('test', 'execute,read,write', 'universe')
""")

args = [self.srv.host, self.srv.args['primary']]
kwargs = { 'user': 'test', 'password': 'test' }
self.con_encoding_utf8 = tarantool.Connection(*args, **kwargs, encoding='utf-8')
self.con_encoding_none = tarantool.Connection(*args, **kwargs, encoding=None)
self.con_encoding_utf8 = tarantool.Connection(*args, encoding='utf-8', **kwargs)
self.con_encoding_none = tarantool.Connection(*args, encoding=None, **kwargs)
self.conns = [self.con_encoding_utf8, self.con_encoding_none]

self.srv.admin("box.schema.create_space('space_str')")
self.srv.admin("""
box.space['space_str']:create_index('primary', {
type = 'tree',
parts = {1, 'str'},
unique = true})
box.space['space_str']:create_index('primary', {
type = 'tree',
parts = {1, 'str'},
unique = true})
""".replace('\n', ' '))

self.srv.admin("box.schema.create_space('space_varbin')")
self.srv.admin(r"""
box.space['space_varbin']:format({
{
'id',
type = 'number',
is_nullable = false
},
{
'varbin',
type = 'varbinary',
is_nullable = false,
}
})
""".replace('\n', ' '))
self.srv.admin("""
box.space['space_varbin']:create_index('id', {
type = 'tree',
parts = {1, 'number'},
unique = true})
""".replace('\n', ' '))
self.srv.admin("""
box.space['space_varbin']:create_index('primary', {
type = 'tree',
parts = {1, 'varbinary'},
unique = true})
box.space['space_varbin']:create_index('varbin', {
type = 'tree',
parts = {2, 'varbinary'},
unique = true})
""".replace('\n', ' '))

self.srv.admin(r"""
local buffer = require('buffer')
local ffi = require('ffi')
function encode_bin(bytes)
local tmpbuf = buffer.ibuf()
local p = tmpbuf:alloc(3 + #bytes)
p[0] = 0x91
p[1] = 0xC4
p[2] = #bytes
for i, c in pairs(bytes) do
p[i + 3 - 1] = c
end
return tmpbuf
end
function bintuple_insert(space, bytes)
local tmpbuf = encode_bin(bytes)
ffi.cdef[[
int box_insert(uint32_t space_id, const char *tuple, const char *tuple_end, box_tuple_t **result);
]]
ffi.C.box_insert(space.id, tmpbuf.rpos, tmpbuf.wpos, nil)
end
""")

self.srv.admin("""
buffer = require('buffer')
ffi = require('ffi')
function encode_bin(bytes)
local tmpbuf = buffer.ibuf()
local p = tmpbuf:alloc(3 + #bytes)
p[0] = 0x91
p[1] = 0xC4
p[2] = #bytes
for i, c in pairs(bytes) do
p[i + 3 - 1] = c
function get_type(arg)
return type(arg)
end
return tmpbuf
end
function bintuple_insert(space, bytes)
local tmpbuf = encode_bin(bytes)
ffi.cdef[[
int box_insert(uint32_t space_id, const char *tuple, const char *tuple_end, box_tuple_t **result);
]]
ffi.C.box_insert(space.id, tmpbuf.rpos, tmpbuf.wpos, nil)
end
""")

def assertNotRaises(self, func, *args, **kwargs):
Expand All @@ -111,50 +107,74 @@ def setUp(self):
if self.srv.is_started():
self.srv.touch_lock()

def test_01_01_string_insert_encoding_utf8_behavior(self):
# encoding = 'utf-8'
#
# Python 3 -> Tarantool -> Python 3
# str -> mp_str (string) -> str
# bytes -> mp_bin (varbinary) -> bytes
def test_01_01_str_encode_for_encoding_utf8_behavior(self):
self.assertNotRaises(
self.con_encoding_utf8.insert,
'space_str', [ 'test_01_01' ])

def test_01_02_string_select_encoding_utf8_behavior(self):
self.adm(r"box.space['space_str']:insert{'test_01_02'}")
def test_01_02_string_decode_for_encoding_utf8_behavior(self):
self.srv.admin(r"box.space['space_str']:insert{'test_01_02'}")

strdata = 'test_01_01'
resp = self.con_encoding_utf8.select('space_str', [strdata])
self.assertEquals(resp[0][0], strdata)
resp = self.con_encoding_utf8.eval("return box.space.space_str:get('test_01_02')")
self.assertSequenceEqual(resp, [['test_01_02']])

@skip_or_run_mp_bin_test
@skip_or_run_varbinary_test
def test_01_03_varbinary_insert_encoding_utf8_behavior(self):
def test_01_03_bytes_encode_for_encoding_utf8_behavior(self):
self.assertNotRaises(
self.con_encoding_utf8.insert,
'space_varbin', [ b'test_01_03' ])
'space_varbin', [ 103, bytes(bytearray.fromhex('DEADBEAF0103')) ])

@skip_or_run_mp_bin_test
@skip_or_run_varbinary_test
def test_01_04_varbinary_select_encoding_utf8_behavior(self):
self.adm(r"""
bintuple_insert(
box.space['space_varbin'],
{0xDE, 0xAD, 0xBE, 0xAF, 0x01, 0x04})
def test_01_04_varbinary_decode_for_encoding_utf8_behavior(self):
self.con_encoding_utf8.execute(r"""
INSERT INTO "space_varbin" VALUES (104, x'DEADBEAF0104');
""")

bindata = bytes(bytearray.fromhex('DEADBEAF0104'))
resp = self.con_encoding_utf8.select('space_varbin', [bindata])
self.assertEquals(resp[0][0], bindata)
resp = self.con_encoding_utf8.execute(r"""
SELECT * FROM "space_varbin" WHERE "varbin" == x'DEADBEAF0104';
""")
self.assertSequenceEqual(resp, [[104, bytes(bytearray.fromhex('DEADBEAF0104'))]])

# encoding = None
#
# Python 3 -> Tarantool -> Python 3
# bytes -> mp_str (string) -> bytes
# str -> mp_str (string)
# mp_bin (string) -> bytes
def test_02_01_str_encode_for_encoding_none_behavior(self):
self.assertNotRaises(
self.con_encoding_none.insert,
'space_str', [ 'test_02_01' ])

def test_02_01_string_insert_encoding_none_behavior(self):
def test_02_02_string_decode_for_encoding_none_behavior(self):
self.srv.admin(r"box.space['space_str']:insert{'test_02_02'}")

resp = self.con_encoding_none.eval("return box.space.space_str:get('test_02_02')")
self.assertSequenceEqual(resp, [[b'test_02_02']])

def test_02_03_bytes_decode_for_encoding_utf8_behavior(self):
self.assertNotRaises(
self.con_encoding_none.insert,
'space_str',
[ bytes(bytearray.fromhex('DEADBEAF0201')) ])
'space_str', [ b'test_02_03' ])

def test_02_02_string_select_encoding_none_behavior(self):
self.adm(r"box.space['space_str']:insert{'\xDE\xAD\xBE\xAF\x02\x02'}")
@skip_or_run_mp_bin_test
@skip_or_run_varbinary_test
def test_02_04_varbinary_decode_for_encoding_utf8_behavior(self):
self.con_encoding_utf8.execute(r"""
INSERT INTO "space_varbin" VALUES (204, x'DEADBEAF0204');
""")

bindata = bytes(bytearray.fromhex('DEADBEAF0202'))
resp = self.con_encoding_none.select('space_str', [bindata])
self.assertEquals(resp[0][0], bindata)
resp = self.con_encoding_utf8.execute(r"""
SELECT * FROM "space_varbin" WHERE "varbin" == x'DEADBEAF0204';
""")
self.assertSequenceEqual(resp, [[204, bytes(bytearray.fromhex('DEADBEAF0204'))]])

@classmethod
def tearDownClass(self):
Expand Down

0 comments on commit 4ec6748

Please sign in to comment.