Add a test illustrating python and postgres encoding mismatch

See #348 (comment)

tests/conftest.py

@@ -5,6 +5,7 @@
 
 import psycopg
 from psycopg.conninfo import make_conninfo
+from psycopg import sql
 import psycopg.errors
 import pytest
 
@@ -23,6 +24,38 @@ def datadir() -> pathlib.Path:
     return pathlib.Path(__file__).parent / "data"
 
 
+@pytest.fixture
+def database_factory(postgresql):
+    dbnames = set()
+
+    def createdb(dbname: str, encoding: str, locale: Optional[str] = None) -> None:
+        with psycopg.connect(postgresql.info.dsn, autocommit=True) as conn:
+            qs = sql.SQL(
+                "CREATE DATABASE {dbname} ENCODING {encoding} TEMPLATE template0"
+            ).format(dbname=sql.Identifier(dbname), encoding=sql.Identifier(encoding))
+            if locale:
+                qs = sql.SQL(" ").join(
+                    [
+                        qs,
+                        sql.SQL("LOCALE {locale}").format(
+                            locale=sql.Identifier(locale)
+                        ),
+                    ]
+                )
+            conn.execute(qs)
+        dbnames.add(dbname)
+
+    yield createdb
+
+    with psycopg.connect(postgresql.info.dsn, autocommit=True) as conn:
+        for dbname in dbnames:
+            conn.execute(
+                sql.SQL("DROP DATABASE IF EXISTS {dbname} WITH (FORCE)").format(
+                    dbname=sql.Identifier(dbname)
+                )
+            )
+
+
 @pytest.fixture
 def execute(postgresql):
     """Create a thread and return an execute() function that will run SQL queries in that

tests/test_data.py

@@ -153,6 +153,28 @@ def test_client_encoding(postgresql, encoding: str) -> None:
     )
 
 
+@pytest.mark.parametrize(
+    "pyenc, pgenc",
+    [
+        ("utf8", "UTF8"),
+        pytest.param("ascii", "SQL_ASCII", marks=pytest.mark.xfail),
+    ],
+)
+def test_postgres_and_python_encoding(
+    database_factory, pyenc: str, pgenc: str, data, postgresql
+) -> None:
+    dbname = pyenc
+    database_factory(dbname, encoding=pgenc)
+    with psycopg.connect(
+        postgresql.info.dsn, dbname=dbname, client_encoding="utf-8"
+    ) as conn:
+        conn.execute("SELECT 'encoding'")
+        (running,) = data.pg_get_activities()
+    assert "'encoding'" in running.query
+    assert running.encoding == pgenc
+    assert running.database == dbname
+
+
 def test_filters_dbname(data, execute):
     data_filtered = attr.evolve(data, filters=types.Filters(dbname="temp"))
     execute("SELECT pg_sleep(2)", dbname="template1", autocommit=True)