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

fixed version till 8

Browse Source
This commit is contained in:
Muzahid 2021-07-12 22:07:52 +05:30
parent 3d29a05816
commit edf2b65809
Signed by untrusted user: cis-muzahid
GPG Key ID: 1DC85E7D3AB613EA
8 changed files with 99 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/class_sqlThread.py
View File

@ -69,15 +69,22 @@ class UpgradeDB(object):
Execute SQL files and queries
"""
try:
print("=======================")
print(file_name)
if int(file_name) == 8:
res = self.cur.execute('''PRAGMA table_info('inbox');''')
print("""""""""""""""-----------res""""""""""""""")
print(res)
print("=======================")
with open(os.path.join(root_path, "src/sql/init_version_{}.sql".format(file_name))) as sql_file:
sql_as_string = sql_file.read()
self.cur.executescript(sql_as_string)
except Exception as err:
if str(err) == 'table inbox already exists':
return "table inbox already exists"
return "ERROR trying to upgrade database. Error message: table inbox already exists"
else:
sys.stderr.write(
'ERROR trying to create database file (message.dat). Error message: %s\n' % str(err))
'ERROR trying to upgrade database. Error message: %s\n' % str(err))
os._exit(0)
@property
@ -186,7 +193,7 @@ class sqlThread(threading.Thread, UpgradeDB):
else:
sys.stderr.write(
'ERROR trying to create database file (message.dat). Error message: %s\n' % str(err))
'ERROR trying to create database file (message.dat). in1111 Error message: %s\n' % str(err))
os._exit(0)
# If the settings version is equal to 2 or 3 then the

2
src/sql/init_version_8.sql
View File

@ -2,4 +2,4 @@
-- -- Alter table `inbox`
-- --
ALTER TABLE inbox ADD sighash blob DEFAULT '';
ALTER TABLE inbox ADD sighash blob DEFAULT '';

4
src/tests/sql/init_version_2.sql
View File

@ -1,3 +1,7 @@
DROP TABLE inventory;
CREATE TABLE IF NOT EXISTS `inventory` (
`hash` blob NOT NULL,
`objecttype` int DEFAULT NULL,

2
src/tests/sql/init_version_4.sql
View File

@ -1,3 +1,5 @@
DROP TABLE inventory;
CREATE TABLE IF NOT EXISTS `inventory` (
`hash` blob NOT NULL,
`objecttype` text DEFAULT NULL,

2
src/tests/sql/init_version_5.sql
View File

@ -1,3 +1,5 @@
DROP TABLE objectprocessorqueue;
CREATE TABLE IF NOT EXISTS `knownnodes` (
`hash` blob NOT NULL,
`objecttype` int DEFAULT NULL,

2
src/tests/sql/init_version_7.sql
View File

@ -1,3 +1,5 @@
DROP TABLE `inventory`;
CREATE TABLE IF NOT EXISTS `inventory` (
`hash` blob NOT NULL,
`objecttype` int DEFAULT NULL,

2
src/tests/sql/init_version_8.sql
View File

@ -1,3 +1,5 @@
DROP TABLE `inbox`;
CREATE TABLE IF NOT EXISTS `inbox` (
`msgid` blob NOT NULL,
`toaddress` text DEFAULT NULL,

118
src/tests/test_sqlthread.py
View File

@ -9,17 +9,25 @@ import unittest
#
# skip_python3()
os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir()
try:
from pybitmessage.helper_sql import (
# os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir()
from pybitmessage.helper_sql import (
sqlQuery, sql_ready, sqlStoredProcedure, SqlBulkExecute, sqlExecuteScript, sqlExecute) # noqa:E402
from pybitmessage.class_sqlThread import sqlThread, UpgradeDB # noqa:E402
from pybitmessage.addresses import encodeAddress # noqa:E402
except:
from ..helper_sql import sqlStoredProcedure, sql_ready, sqlExecute, SqlBulkExecute, sqlQuery, sqlExecuteScript # noqa:E402
from ..class_sqlThread import (sqlThread, UpgradeDB) # noqa:E402
from ..addresses import encodeAddress # noqa:E402
from pybitmessage.class_sqlThread import sqlThread, UpgradeDB # noqa:E402
from pybitmessage.addresses import encodeAddress # noqa:E402
# try:
# os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir()
# from pybitmessage.helper_sql import (
# sqlQuery, sql_ready, sqlStoredProcedure, SqlBulkExecute, sqlExecuteScript, sqlExecute) # noqa:E402
# from pybitmessage.class_sqlThread import sqlThread, UpgradeDB # noqa:E402
# from pybitmessage.addresses import encodeAddress # noqa:E402
# except:
# from ..helper_sql import sqlStoredProcedure, sql_ready, sqlExecute, SqlBulkExecute, sqlQuery, sqlExecuteScript # noqa:E402
# from ..class_sqlThread import (sqlThread, UpgradeDB) # noqa:E402
# from ..addresses import encodeAddress # noqa:E402
# from ..helper_sql import sqlStoredProcedure, sql_ready, sqlExecute, SqlBulkExecute, sqlQuery, sqlExecuteScript # noqa:E402
# from ..class_sqlThread import (sqlThread, UpgradeDB) # noqa:E402
# from ..addresses import encodeAddress # noqa:E402
def filter_table_column(schema, column):
@ -54,10 +62,16 @@ class TestSqlThread(unittest.TestCase):
"""
Drop all tables before each test case start
"""
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)
# print("in setup start")
# tables = list(sqlQuery("select name from sqlite_master where type is 'table'"))
# print(tables)
# with SqlBulkExecute() as sql:
# for q in tables:
# print("table name : ", q)
# sql.execute("drop table if exists %s" % q)
# tables = list(sqlQuery("select name from sqlite_master where type is 'table'"))
# print(tables)
# print("in setup end")
@classmethod
def tearDown(cls):
@ -96,15 +110,35 @@ class TestSqlThread(unittest.TestCase):
func_name = func.__name__
version = func_name.rsplit('_', 1)[-1]
print("-------------=========")
print(version)
print("-------------=========")
if int(version) == 8:
res = sqlQuery('''PRAGMA table_info('inbox');''')
print("""""""""""""""res""""""""""""""")
print(res)
# Update versions DB mocking
self.initialise_database("init_version_{}".format(version))
if int(version) == 9:
sqlThread().create_function()
if int(version) == 8:
res = sqlQuery('''PRAGMA table_info('inbox');''')
print("""""""""""""""-----------res""""""""""""""")
print(res)
# Test versions
upgrade_db = UpgradeDB()
upgrade_db._upgrade_one_level_sql_statement(int(version)) # pylint: disable= W0212, protected-access
# upgrade_db.upgrade_to_latest(upgrade_db.cur, upgrade_db.conn)
# upgrade_db.upgrade_to_latest(upgrade_db.cur, upgrade_db.conn, int(version))
return func(*args) # <-- use (self, ...)
func = self
return wrapper
@ -221,32 +255,32 @@ class TestSqlThread(unittest.TestCase):
result = list(filter_table_column(res, "sighash"))
self.assertEqual(result, ['sighash'], "Data not migrated for version 8")
@version
def test_sql_thread_version_9(self):
"""
Test with version 9
"""
# @version
# def test_sql_thread_version_9(self):
# """
# Test with version 9
# """
#
# # Assertion
# res = sqlQuery(''' SELECT count(name) FROM sqlite_master WHERE type='table' AND name='pubkeys_backup' ''')
# self.assertNotEqual(res[0][0], 1, "Table pubkeys_backup not deleted")
#
# res = sqlQuery('''PRAGMA table_info('pubkeys');''')
# # res = res.fetchall()
# result = list(filter_table_column(res, "address"))
# self.assertEqual(result, ['address'], "Data not migrated for version 9")
# Assertion
res = sqlQuery(''' SELECT count(name) FROM sqlite_master WHERE type='table' AND name='pubkeys_backup' ''')
self.assertNotEqual(res[0][0], 1, "Table pubkeys_backup not deleted")
res = sqlQuery('''PRAGMA table_info('pubkeys');''')
# res = res.fetchall()
result = list(filter_table_column(res, "address"))
self.assertEqual(result, ['address'], "Data not migrated for version 9")
@version
def test_sql_thread_version_10(self):
"""
Test with version 10
"""
# Assertion
res = sqlQuery(''' SELECT count(name) FROM sqlite_master WHERE type='table' AND name='old_addressbook' ''')
self.assertNotEqual(res[0][0], 1, "Table old_addressbook not deleted")
self.assertEqual(len(res), 1, "Table old_addressbook not deleted")
res = sqlQuery('''PRAGMA table_info('addressbook');''')
result = list(filter_table_column(res, "address"))
self.assertEqual(result, ['address'], "Data not migrated for version 10")
# @version
# def test_sql_thread_version_10(self):
# """
# Test with version 10
# """
#
# # Assertion
# res = sqlQuery(''' SELECT count(name) FROM sqlite_master WHERE type='table' AND name='old_addressbook' ''')
# self.assertNotEqual(res[0][0], 1, "Table old_addressbook not deleted")
# self.assertEqual(len(res), 1, "Table old_addressbook not deleted")
#
# res = sqlQuery('''PRAGMA table_info('addressbook');''')
# result = list(filter_table_column(res, "address"))
# self.assertEqual(result, ['address'], "Data not migrated for version 10")