test/python/test_db_async_connection.py

   1 """
   2 Tests for function providing a non-blocking query interface towards PostgreSQL.
   3 """
   4 from contextlib import closing
   5 import concurrent.futures
   6
   7 import pytest
   8 import psycopg2
   9 from psycopg2.extras import wait_select
  10
  11 from nominatim.db.async_connection import DBConnection, DeadlockHandler
  12
  13
  14 @pytest.fixture
  15 def conn(temp_db):
  16     with closing(DBConnection('dbname=' + temp_db)) as c:
  17         yield c
  18
  19
  20 @pytest.fixture
  21 def simple_conns(temp_db):
  22     conn1 = psycopg2.connect('dbname=' + temp_db)
  23     conn2 = psycopg2.connect('dbname=' + temp_db)
  24
  25     yield conn1.cursor(), conn2.cursor()
  26
  27     conn1.close()
  28     conn2.close()
  29
  30
  31 def test_simple_query(conn, temp_db_conn):
  32     conn.connect()
  33
  34     conn.perform('CREATE TABLE foo (id INT)')
  35     conn.wait()
  36
  37     temp_db_conn.table_exists('foo')
  38
  39
  40 def test_wait_for_query(conn):
  41     conn.connect()
  42
  43     conn.perform('SELECT pg_sleep(1)')
  44
  45     assert not conn.is_done()
  46
  47     conn.wait()
  48
  49
  50 def test_bad_query(conn):
  51     conn.connect()
  52
  53     conn.perform('SELECT efasfjsea')
  54
  55     with pytest.raises(psycopg2.ProgrammingError):
  56         conn.wait()
  57
  58
  59 def exec_with_deadlock(cur, sql, detector):
  60     with DeadlockHandler(lambda *args: detector.append(1)):
  61         cur.execute(sql)
  62
  63
  64 def test_deadlock(simple_conns):
  65     print(psycopg2.__version__)
  66     cur1, cur2 = simple_conns
  67
  68     cur1.execute("""CREATE TABLE t1 (id INT PRIMARY KEY, t TEXT);
  69                     INSERT into t1 VALUES (1, 'a'), (2, 'b')""")
  70     cur1.connection.commit()
  71
  72     cur1.execute("UPDATE t1 SET t = 'x' WHERE id = 1")
  73     cur2.execute("UPDATE t1 SET t = 'x' WHERE id = 2")
  74
  75     # This is the tricky part of the test. The first SQL command runs into
  76     # a lock and blocks, so we have to run it in a separate thread. When the
  77     # second deadlocking SQL statement is issued, Postgresql will abort one of
  78     # the two transactions that cause the deadlock. There is no way to tell
  79     # which one of the two. Therefore wrap both in a DeadlockHandler and
  80     # expect that exactly one of the two triggers.
  81     with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
  82         deadlock_check = []
  83         try:
  84             future = executor.submit(exec_with_deadlock, cur2,
  85                                      "UPDATE t1 SET t = 'y' WHERE id = 1",
  86                                      deadlock_check)
  87
  88             while not future.running():
  89                 pass
  90
  91
  92             exec_with_deadlock(cur1, "UPDATE t1 SET t = 'y' WHERE id = 2",
  93                                deadlock_check)
  94         finally:
  95             # Whatever happens, make sure the deadlock gets resolved.
  96             cur1.connection.rollback()
  97
  98         future.result()
  99
 100         assert len(deadlock_check) == 1
 101
 102