Second test with pytest-postgresql still using schema from first test

Question

I am trying to perform some tests using pytest-postgresql but it seems I do not fully grasp how the engine generated from the fixture works. The code below is the minimal example I could write to reproduce the error I am getting. There are 2 tests named for tokyo and london and each is meant to use the DB connection to create a schema with the name of the city and then to create the table citymeta in the schema.

import pytest
from pytest_postgresql.janitor import DatabaseJanitor
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import Session
from sqlalchemy import text
from sqlalchemy import (Column, Integer, String)
from sqlalchemy.orm import declarative_base


Base = declarative_base()


class BaseModel(Base):
    __abstract__ = True
    id = Column(Integer, primary_key=True)


class CityMeta(BaseModel):
    __tablename__ = "citymeta"
    id = Column(Integer, primary_key=True)
    version = Column(String(10), nullable=False)


def create_table(engine, schema):
    with engine.connect() as conn:
        conn.execute(text(f"""CREATE SCHEMA IF NOT EXISTS
                                {schema}"""))
        conn.commit()

    for table in BaseModel.metadata.tables.values():
        table.schema = schema
        table.create(engine, checkfirst=True)


schema1 = "tokyo"
schema2 = "london"

@pytest.fixture(scope="session")
def engine_postgresql(postgresql_proc):
    with DatabaseJanitor(
        postgresql_proc.user,
        postgresql_proc.host,
        postgresql_proc.port,
        postgresql_proc.dbname,
        postgresql_proc.version,
        password=postgresql_proc.password,
    ):
        yield create_engine(
            f"postgresql+psycopg2://{postgresql_proc.user}:"
            f"{postgresql_proc.password}@{postgresql_proc.host}:"
            f"{postgresql_proc.port}/{postgresql_proc.dbname}"
        )


def test_tokyo(engine_postgresql):
    create_table(engine_postgresql, schema1)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema1)

    with Session(engine_postgresql) as session:
        meta = CityMeta(id=1, version='2.3')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '2.3'


def test_london(engine_postgresql):
    create_table(engine_postgresql, schema2)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema2)

    with Session(engine_postgresql) as session:
        meta = CityMeta(id=1, version='0.0')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '0.0'

Since each test creates a separate schema, I would expect both the tests to run without an issue. However, I am getting the following error at the second test:

 sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "citymeta_pkey"
E       DETAIL:  Key (id)=(1) already exists.
E       
E       [SQL: INSERT INTO tokyo.citymeta (id, version) VALUES (%(id)s, %(version)s)]
E       [parameters: {'id': 1, 'version': '0.0'}]
E       (Background on this error at: https://sqlalche.me/e/20/gkpj)

It seems that during the second test (london), even though the london schema is created successfully, when the session is started, somehow it still uses the schema tokyo, and it cannot insert the data due to key violations (both ids are 1). I know that pytest-postgresql maintains the same database until the end of the tests, but, since a new schema is generated in the second test, shouldn't the new schema be used by the new session?

Answer 1

Inspired by the answers to this question I managed to make the test run by adding this line at the second test:

engine_postgresql.update_execution_options(schema_translate_map={schema1: schema2})