Add butler test for postgres registry

This will ensure that datastore is tested with postgres.

tests/test_butler.py

@@ -22,6 +22,7 @@
 """Tests for Butler.
 """
 
+import gc
 import logging
 import os
 import pathlib
@@ -49,13 +50,24 @@ def mock_s3(cls):
         return cls
 
 
+try:
+    # It's possible but silly to have testing.postgresql installed without
+    # having the postgresql server installed (because then nothing in
+    # testing.postgresql would work), so we use the presence of that module
+    # to test whether we can expect the server to be available.
+    import testing.postgresql
+except ImportError:
+    testing = None
+
+
 try:
     from cheroot import wsgi
     from wsgidav.wsgidav_app import WsgiDAVApp
 except ImportError:
     WsgiDAVApp = None
 
 import astropy.time
+import sqlalchemy
 from lsst.daf.butler import (
     Butler,
     ButlerConfig,
@@ -1486,6 +1498,64 @@ def testPytypeCoercion(self):
             butler.get(datasetTypeName, dataId=dataId)
 
 
+@unittest.skipUnless(testing is not None, "testing.postgresql module not found")
+class PostgresPosixDatastoreButlerTestCase(FileDatastoreButlerTests, unittest.TestCase):
+    """PosixDatastore specialization of a butler using Postgres"""
+
+    configFile = os.path.join(TESTDIR, "config/basic/butler.yaml")
+    fullConfigKey = ".datastore.formatters"
+    validationCanFail = True
+    datastoreStr = ["/tmp"]
+    datastoreName = [f"FileDatastore@{BUTLER_ROOT_TAG}"]
+    registryStr = "PostgreSQL@test"
+
+    @staticmethod
+    def _handler(postgresql):
+        engine = sqlalchemy.engine.create_engine(postgresql.url())
+        with engine.begin() as connection:
+            connection.execute(sqlalchemy.text("CREATE EXTENSION btree_gist;"))
+
+    @classmethod
+    def setUpClass(cls):
+        # Create the postgres test server.
+        cls.postgresql = testing.postgresql.PostgresqlFactory(
+            cache_initialized_db=True, on_initialized=cls._handler
+        )
+        super().setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        # Clean up any lingering SQLAlchemy engines/connections
+        # so they're closed before we shut down the server.
+        gc.collect()
+        cls.postgresql.clear_cache()
+        super().tearDownClass()
+
+    def setUp(self):
+        self.server = self.postgresql()
+
+        # Need to add a registry section to the config.
+        self._temp_config = False
+        config = Config(self.configFile)
+        config["registry", "db"] = self.server.url()
+        with tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) as fh:
+            config.dump(fh)
+            self.configFile = fh.name
+            self._temp_config = True
+        super().setUp()
+
+    def tearDown(self):
+        self.server.stop()
+        if self._temp_config and os.path.exists(self.configFile):
+            os.remove(self.configFile)
+        super().tearDown()
+
+    def testMakeRepo(self):
+        # The base class test assumes that it's using sqlite and assumes
+        # the config file is acceptable to sqlite.
+        raise unittest.SkipTest("Postgres config is not compatible with this test.")
+
+
 class InMemoryDatastoreButlerTestCase(ButlerTests, unittest.TestCase):
     """InMemoryDatastore specialization of a butler"""