Commit 424a3751 authored by Lukáš Lalinský's avatar Lukáš Lalinský

Add a test with schema support

See https://github.com/lalinsky/python-phoenixdb/issues/3
parent b0be0542
Pipeline #19636 passed with stages
in 10 minutes and 24 seconds
import unittest
import phoenixdb
import phoenixdb.cursor
from phoenixdb.errors import InternalError
from phoenixdb.tests import TEST_DB_URL
......@@ -63,3 +64,21 @@ class PhoenixDatabaseTest(unittest.TestCase):
with db.cursor(cursor_factory=phoenixdb.cursor.DictCursor) as cursor:
self._check_dict_cursor(cursor)
def test_schema(self):
db = phoenixdb.connect(TEST_DB_URL, autocommit=True)
self.addCleanup(db.close)
with db.cursor() as cursor:
try:
cursor.execute("CREATE SCHEMA IF NOT EXISTS test_schema")
except InternalError as e:
if "phoenix.schema.isNamespaceMappingEnabled" in e.message:
self.skipTest(e.message)
raise
cursor.execute("DROP TABLE IF EXISTS test_schema.test")
cursor.execute("CREATE TABLE test_schema.test (id INTEGER PRIMARY KEY, text VARCHAR)")
cursor.execute("UPSERT INTO test_schema.test VALUES (?, ?)", [1, 'text 1'])
cursor.execute("SELECT * FROM test_schema.test ORDER BY id")
self.assertEqual(cursor.fetchall(), [[1, 'text 1']])
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment