1  """Tests for SQL thread"""
 2  # flake8: noqa:E402
 3  import os
 4  import tempfile
 5  import threading
 6  import unittest
 7  
 8  from .common import skip_python3
 9  
10  skip_python3()
11  
12  os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir()
13  
14  from pybitmessage.helper_sql import (
15      sqlQuery, sql_ready, sqlStoredProcedure)  # noqa:E402
16  from pybitmessage.class_sqlThread import sqlThread  # noqa:E402
17  from pybitmessage.addresses import encodeAddress  # noqa:E402
18  
19  
20  class TestSqlThread(unittest.TestCase):
21      """Test case for SQL thread"""
22  
23      @classmethod
24      def setUpClass(cls):
25          # Start SQL thread
26          sqlLookup = sqlThread()
27          sqlLookup.daemon = True
28          sqlLookup.start()
29          sql_ready.wait()
30  
31      @classmethod
32      def tearDownClass(cls):
33          sqlStoredProcedure('exit')
34          for thread in threading.enumerate():
35              if thread.name == "SQL":
36                  thread.join()
37  
38      def test_create_function(self):
39          """Check the result of enaddr function"""
40          encoded_str = encodeAddress(4, 1, "21122112211221122112")
41  
42          query = sqlQuery('SELECT enaddr(4, 1, "21122112211221122112")')
43          self.assertEqual(
44              query[0][-1], encoded_str, "test case fail for create_function")