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:
<code>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>
<code>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>
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
<code>docker run -d --rm --name postgres-container -e POSTGRES_USER=testuser-e POSTGRES_PASSWORD=examplepass -p 5432:5432 postgres:latest
</code>
<code>docker run -d --rm --name postgres-container -e POSTGRES_USER=testuser-e POSTGRES_PASSWORD=examplepass -p 5432:5432 postgres:latest
</code>
docker run -d --rm --name postgres-container -e POSTGRES_USER=testuser-e POSTGRES_PASSWORD=examplepass -p 5432:5432 postgres:latest
pytest.ini
<code>[pytest]
verbosity_assertions = 1
env =
DB_PASS=examplepass
DB_USER=testuser
DB_HOST=localhost
DB_NAME=testdb
DB_PORT=5432
</code>
<code>[pytest]
verbosity_assertions = 1
env =
DB_PASS=examplepass
DB_USER=testuser
DB_HOST=localhost
DB_NAME=testdb
DB_PORT=5432
</code>
[pytest]
verbosity_assertions = 1
env =
DB_PASS=examplepass
DB_USER=testuser
DB_HOST=localhost
DB_NAME=testdb
DB_PORT=5432
conftest.py setup
<code>@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
</code>
<code>@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
</code>
@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
<code>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')"))
</code>
<code>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')"))
</code>
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
<code>INFO root:conftest.py:47 postgresql://testuser:examplepass@localhost/testdb
INFO root:conftest.py:49 postgresql://testuser:***@localhost/testdb
</code>
<code>INFO root:conftest.py:47 postgresql://testuser:examplepass@localhost/testdb
INFO root:conftest.py:49 postgresql://testuser:***@localhost/testdb
</code>
INFO root:conftest.py:47 postgresql://testuser:examplepass@localhost/testdb
INFO root:conftest.py:49 postgresql://testuser:***@localhost/testdb