pythonsqlalchemypytestwindows-subsystem-for-linuxalembic

Authentication fails on localhost postgresql with alembic and pytest


I want to connect to a postgresql database via alembic in pytest. I can connect to the database via pg admin with the password i set, but i always get the error:

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at "localhost" (127.0.0.1), port 5432 failed: FATAL:  password authentication failed for user "testuser"

Code is in VisualStudio code and opened in WSL.

docker postgresql db

docker run -d --rm --name postgres-container -e POSTGRES_USER=testuser -e POSTGRES_PASSWORD=examplepass -p 5432:5432 postgres:latest

pytest.ini

[pytest]
verbosity_assertions = 1
env = 
    DB_PASS=examplepass
    DB_USER=testuser
    DB_HOST=localhost
    DB_NAME=testdb
    DB_PORT=5432

conftest.py setup

@pytest.fixture(scope='module')
def connection_string():
    return f'postgresql://{os.getenv("DB_USER")}:{os.getenv("DB_PASS")}@{os.getenv("DB_HOST")}/{os.getenv("DB_NAME")}'

@pytest.fixture(scope='module')
def engine(connection_string):
    logging.info(connection_string)
    engine = create_engine(connection_string)
    logging.info(engine.url)
    return engine

@pytest.fixture(scope='module')
def tables(engine):
    alembic_cfg = AlembicConfig("alembic.ini")
    alembic_cfg.set_main_option('sqlalchemy.url', str(engine.url))
    command.upgrade(alembic_cfg, "head")
    yield
    command.downgrade(alembic_cfg, "base")

@pytest.fixture(scope='module')
def session(engine, tables, context):
    """Returns an sqlalchemy session, and after the test tears down everything properly."""
    connection = engine.connect()
    Session = sessionmaker(bind=connection)
    session = Session()
    # inserts example tables
    insert_into_tables(session, context)
    yield session

test_file.py

sys.path.append(".")
from sqlalchemy import text

def test_db_setup(session):
    # Try to execute a simple query to check if a table exists
    result_table_exists = session.execute(text("SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'task')"))

logging output shows the correct URL

INFO    root:conftest.py:47 postgresql://testuser:examplepass@localhost/testdb
INFO    root:conftest.py:49 postgresql://testuser:***@localhost/testdb

Solution

  • Problem was the

    alembic_cfg.set_main_option('sqlalchemy.url', str(engine.url))
    

    As the password is encrypted, it needs the connection_string instead to authorize correctly