Bitmessage/PyBitmessage-2024-12-01
Archived
2
0
Fork 0
Code Issues 168 Pull Requests 53 Actions Packages Projects Releases 5 Wiki Activity

Remove bikeshed from the new sqlthread test case

Browse Source
This commit is contained in:
Dmitri Bogomolov 2021-05-02 19:42:10 +03:00
parent 583ed2151d
commit b1a0a54d56
Signed by untrusted user: g1itch
GPG Key ID: 720A756F18DEED13
2 changed files with 15 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/tests/sql/create_function.sql
View File

@ -1,9 +0,0 @@
CREATE TABLE `testhash` (
`addressversion` int DEFAULT NULL,
`hash` blob DEFAULT NULL,
`address` text DEFAULT NULL,
UNIQUE(address) ON CONFLICT IGNORE
);
INSERT INTO testhash (addressversion, hash) VALUES(4, "21122112211221122112");

66
src/tests/test_sqlthread.py
View File

@ -1,24 +1,19 @@
""" """Tests for SQL thread"""
Test for sqlThread
"""
import os import os
import unittest import tempfile
from ..helper_sql import sqlStoredProcedure, sql_ready, sqlExecute, SqlBulkExecute, sqlQuery, sqlExecuteScript
from ..class_sqlThread import (sqlThread)
from ..addresses import encodeAddress
import threading import threading
import unittest
os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir() # noqa:E402
from pybitmessage.helper_sql import sqlQuery, sql_ready, sqlStoredProcedure
from pybitmessage.class_sqlThread import sqlThread
from pybitmessage.addresses import encodeAddress
class TestSqlThread(unittest.TestCase): class TestSqlThread(unittest.TestCase):
""" """Test case for SQL thread"""
Test case for SQLThread
"""
# query file path
root_path = os.path.dirname(os.path.dirname(__file__))
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
@ -28,18 +23,6 @@ class TestSqlThread(unittest.TestCase):
sqlLookup.start() sqlLookup.start()
sql_ready.wait() sql_ready.wait()
@classmethod
def setUp(cls):
tables = list(sqlQuery("select name from sqlite_master where type is 'table'"))
with SqlBulkExecute() as sql:
for q in tables:
sql.execute("drop table if exists %s" % q)
@classmethod
def tearDown(cls):
pass
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
sqlStoredProcedure('exit') sqlStoredProcedure('exit')
@ -47,29 +30,10 @@ class TestSqlThread(unittest.TestCase):
if thread.name == "SQL": if thread.name == "SQL":
thread.join() thread.join()
def initialise_database(self, file):
"""
Initialise DB
"""
with open(os.path.join(self.root_path, "tests/sql/{}.sql".format(file)), 'r') as sql_as_string:
# sql_as_string = open(os.path.join(self.root_path, "tests/sql/{}.sql".format(file))).read()
sql_as_string = sql_as_string.read()
sqlExecuteScript(sql_as_string)
def test_create_function(self): def test_create_function(self):
# call create function """Check the result of enaddr function"""
encoded_str = encodeAddress(4, 1, "21122112211221122112") encoded_str = encodeAddress(4, 1, "21122112211221122112")
# Initialise Database query = sqlQuery('SELECT enaddr(4, 1, "21122112211221122112")')
self.initialise_database("create_function") self.assertEqual(
query[0][-1], encoded_str, "test case fail for create_function")
sqlExecute('''INSERT INTO testhash (addressversion, hash) VALUES(4, "21122112211221122112")''')
# call function in query
# sqlExecute('''UPDATE testhash SET address=(enaddr(testhash.addressversion, 1, hash)) WHERE hash=testhash.hash''')
sqlExecute('''UPDATE testhash SET address=(enaddr(testhash.addressversion, 1, hash));''')
# Assertion
query = sqlQuery('''select * from testhash;''')
self.assertEqual(query[0][-1], encoded_str, "test case fail for create_function")
sqlExecute('''DROP TABLE testhash''')