Add tests for swh.core.db.

conftest.py

+from hypothesis import settings
+
+# define tests profile. Full documentation is at:
+# https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles
+settings.register_profile("fast", max_examples=5, deadline=5000)
+settings.register_profile("slow", max_examples=20, deadline=5000)

requirements-test.txt

 pytest < 4
 pytest-postgresql
 requests-mock
+hypothesis >= 3.11.0

swh/core/db/__init__.py

@@ -94,13 +94,11 @@ class BaseDb:
 
         """
         conn = psycopg2.connect(*args, **kwargs)
-        cls.adapt_conn(conn)
         return cls(conn)
 
     @classmethod
     def from_pool(cls, pool):
         conn = pool.getconn()
-        cls.adapt_conn(conn)
         return cls(conn, pool=pool)
 
     def __init__(self, conn, pool=None):
@@ -111,6 +109,7 @@ class BaseDb:
             pool: psycopg2 pool of connections
 
         """
+        self.adapt_conn(conn)
         self.conn = conn
         self.pool = pool
 
@@ -153,7 +152,7 @@ class BaseDb:
         """Copy items' entries to table tblname with columns information.
 
         Args:
-            items (dict): dictionary of data to copy over tblname.
+            items (List[dict]): dictionaries of data to copy over tblname.
             tblname (str): destination table's name.
             columns ([str]): keys to access data in items and also the
               column names in the destination table.

swh/core/tests/test_db.py

+# Copyright (C) 2019  The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+import os.path
+import tempfile
+import unittest
+
+from hypothesis import strategies, given
+import pytest
+
+from swh.core.db import BaseDb
+from swh.core.tests.db_testing import SingleDbTestFixture
+
+
+INIT_SQL = '''
+create table test_table
+(
+    i       int,
+    txt     text,
+    bytes   bytea
+);
+'''
+
+db_rows = strategies.lists(strategies.tuples(
+    strategies.integers(-2147483648, +2147483647),
+    strategies.text(
+        alphabet=strategies.characters(
+            blacklist_categories=['Cs'],  # surrogates
+            blacklist_characters=[
+                '\x00',  # pgsql does not support the null codepoint
+                '\r',  # pgsql normalizes those
+            ]
+        ),
+    ),
+    strategies.binary(),
+))
+
+
+@pytest.mark.db
+class TestDb(SingleDbTestFixture, unittest.TestCase):
+    TEST_DB_NAME = 'test-db'
+
+    @classmethod
+    def setUpClass(cls):
+        with tempfile.TemporaryDirectory() as td:
+            with open(os.path.join(td, 'init.sql'), 'a') as fd:
+                fd.write(INIT_SQL)
+
+            cls.TEST_DB_DUMP = os.path.join(td, '*.sql')
+
+            super().setUpClass()
+
+    def setUp(self):
+        super().setUp()
+        self.db = BaseDb(self.conn)
+
+    def test_initialized(self):
+        cur = self.db.cursor()
+        cur.execute("insert into test_table values (1, %s, %s);",
+                    ('foo', b'bar'))
+        cur.execute("select * from test_table;")
+        self.assertEqual(list(cur), [(1, 'foo', b'bar')])
+
+    def test_reset_tables(self):
+        cur = self.db.cursor()
+        cur.execute("insert into test_table values (1, %s, %s);",
+                    ('foo', b'bar'))
+        self.reset_db_tables('test-db')
+        cur.execute("select * from test_table;")
+        self.assertEqual(list(cur), [])
+
+    @given(db_rows)
+    def test_copy_to(self, data):
+        # the table is not reset between runs by hypothesis
+        self.reset_db_tables('test-db')
+
+        items = [dict(zip(['i', 'txt', 'bytes'], item)) for item in data]
+        self.db.copy_to(items, 'test_table', ['i', 'txt', 'bytes'])
+
+        cur = self.db.cursor()
+        cur.execute('select * from test_table;')
+        self.assertCountEqual(list(cur), data)

tox.ini

@@ -5,8 +5,18 @@ envlist=flake8,py3
 deps =
   .[testing]
   pytest-cov
+  pifpaf
 commands =
-  pytest --cov=swh --cov-branch {posargs}
+  pifpaf run postgresql -- pytest --hypothesis-profile=fast --cov=swh --cov-branch {posargs}
+
+[testenv:py3-slow]
+deps =
+  .[testing]
+  pytest-cov
+  pifpaf
+commands =
+  pifpaf run postgresql -- pytest --hypothesis-profile=slow --cov=swh --cov-branch {posargs}
+
 
 [testenv:flake8]
 skip_install = true