Tuuli/based
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Changed project structure for it to build correctly

Browse source
This commit is contained in:
Andrew 2023-03-12 00:46:59 +07:00
parent a533dddf1a
commit 6858eeaebe
11 changed files with 45 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

174
tests/db_test.py Normal file
View file

@ -0,0 +1,174 @@
import unittest
import uuid
from based.db import DBConnector, ColumnUpdate, ColumnCondition
from based.columns import *
conninfo = "postgresql://postgres:asarch6122@localhost"
connector = DBConnector(conninfo)
class TestDBConnector(unittest.TestCase):
def test_connection(self):
with connector.connection() as conn:
one: dict = conn.execute('SELECT 1 as "ONE"').fetchone() # type: ignore
self.assertIsNotNone(one)
self.assertEqual(one["ONE"], 1)
def test_columnDefinition(self):
with connector.connection() as conn:
cdef1 = PrimarySerialColumnDefinition("id")
cdef2 = PrimaryUUIDColumnDefinition("uid")
cdef3 = TextColumnDefinition("name")
cdef4 = TextColumnDefinition("name", "John Doe")
self.assertEqual(cdef1.sql().as_string(conn), '"id" SERIAL PRIMARY KEY')
self.assertEqual(
cdef2.sql().as_string(conn),
'"uid" uuid DEFAULT gen_random_uuid() PRIMARY KEY',
)
self.assertEqual(cdef3.sql().as_string(conn), '"name" TEXT')
self.assertEqual(
cdef4.sql().as_string(conn), "\"name\" TEXT DEFAULT 'John Doe'"
)
def test_tableCreation(self):
connector.dropTable("test_table")
self.assertFalse(connector.tableExists("test_table"))
connector.createTable(
"test_table",
[
PrimarySerialColumnDefinition("id"),
TextColumnDefinition("name", "John Doe"),
],
)
self.assertTrue(connector.tableExists("test_table"))
connector.dropTable("test_table")
self.assertFalse(connector.tableExists("test_table"))
def test_differentColumnsWorking(self):
connector.dropTable("test_table")
self.assertFalse(connector.tableExists("test_table"))
connector.createTable(
"test_table",
[
PrimarySerialColumnDefinition("id"),
TextColumnDefinition("textcol", "John Doe"),
BigintColumnDefinition("bigintcol", 2**30),
BooleanColumnDefinition("boolcol", True),
DateColumnDefinition("datecol", date.today()),
TimestampColumnDefinition("tscol", date.today()),
DoubleColumnDefinition("doublecol", 3.14),
IntegerColumnDefinition("intcol", 100),
UUIDColumnDefinition("uuidcol", "00000000-0000-0000-0000-000000000000"),
],
)
self.assertTrue(connector.tableExists("test_table"))
connector.insertIntoTable("test_table", {})
connector.insertIntoTable(
"test_table",
{
"textcol": "Jane Doe",
"bigintcol": 3**30,
"boolcol": False,
"datecol": date.today(),
"tscol": date.today(),
"doublecol": 3.14,
"intcol": 100,
"uuidcol": uuid.uuid4(),
},
)
res = connector.selectFromTable("test_table", ["*"])
self.assertEqual(len(res), 2)
self.assertEqual(res[0]["textcol"], "John Doe")
self.assertEqual(res[0]["bigintcol"], 2**30)
self.assertEqual(res[0]["boolcol"], True)
self.assertIsNotNone(res[0]["datecol"])
self.assertIsNotNone(res[0]["tscol"])
self.assertEqual(res[0]["doublecol"], 3.14)
self.assertEqual(res[0]["intcol"], 100)
self.assertEqual(
res[0]["uuidcol"], uuid.UUID("00000000-0000-0000-0000-000000000000")
)
self.assertEqual(res[1]["textcol"], "Jane Doe")
self.assertEqual(res[1]["bigintcol"], 3**30)
self.assertEqual(res[1]["boolcol"], False)
self.assertIsNotNone(res[1]["datecol"])
self.assertIsNotNone(res[1]["tscol"])
self.assertEqual(res[1]["doublecol"], 3.14)
self.assertEqual(res[1]["intcol"], 100)
self.assertIsNotNone(res[1]["uuidcol"])
connector.dropTable("test_table")
self.assertFalse(connector.tableExists("test_table"))
def test_tableInsertSelect(self):
connector.dropTable("test_table")
connector.createTable(
"test_table",
[
PrimarySerialColumnDefinition("id"),
TextColumnDefinition("name", "John Doe"),
],
)
connector.insertIntoTable("test_table", {"name": "John Doe"})
connector.insertIntoTable("test_table", {"name": "Jane Doe"})
connector.insertIntoTable("test_table", {"name": "John Smith"})
connector.insertIntoTable("test_table", {"name": "Jane Smith"})
rows = connector.selectFromTable("test_table", ["name"])
self.assertEqual(len(rows), 4)
self.assertEqual(rows[0]["name"], "John Doe")
self.assertEqual(rows[1]["name"], "Jane Doe")
self.assertEqual(rows[2]["name"], "John Smith")
self.assertEqual(rows[3]["name"], "Jane Smith")
connector.dropTable("test_table")
def test_tableUpdate(self):
connector.dropTable("test_table")
connector.createTable(
"test_table",
[
PrimarySerialColumnDefinition("id"),
TextColumnDefinition("name", "John Doe"),
],
)
connector.insertIntoTable("test_table", {"name": "John Doe"})
rows = connector.selectFromTable("test_table", ["name"])
self.assertEqual(rows[0]["name"], "John Doe")
connector.updateDataInTable(
"test_table",
[
ColumnUpdate("name", "John"),
],
[
ColumnCondition("name", "John Doe"),
],
)
rows = connector.selectFromTable("test_table", ["name"])
self.assertEqual(rows[0]["name"], "John")
connector.dropTable("test_table")
def test_tableDeleteFrom(self):
connector.dropTable("test_table")
connector.createTable(
"test_table",
[
PrimarySerialColumnDefinition("id"),
TextColumnDefinition("name", "John Doe"),
],
)
connector.insertIntoTable("test_table", {"name": "John Doe"})
connector.insertIntoTable("test_table", {"name": "Jane Doe"})
connector.insertIntoTable("test_table", {"name": "Mikhail Prokopenko"})
rows = connector.selectFromTable("test_table", ["name"])
self.assertEqual(len(rows), 3)
self.assertEqual(rows[0]["name"], "John Doe")
self.assertEqual(rows[1]["name"], "Jane Doe")
self.assertEqual(rows[2]["name"], "Mikhail Prokopenko")
connector.deleteFromTable("test_table", [ColumnCondition("name", "John Doe")])
rows = connector.selectFromTable("test_table", ["name"])
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0]["name"], "Jane Doe")
self.assertEqual(rows[1]["name"], "Mikhail Prokopenko")
connector.dropTable("test_table")
if __name__ == "__main__":
unittest.main()