Can't connect to PostgreSQL inside docker container with SQLAlchemy when running pytest

Question

I am creating FastAPI application and trying to setup two database containers with docker (one db is for testing). I successfully connect to the first database, but for some reason can't connect to the second one. Here is docker-compose.yml

services:

  # FastAPI app
  web:
    build: .
    command: bash -c  'alembic upgrade head && uvicorn src.main:app --host 0.0.0.0 --port 8000 --reload'
    volumes:
      - .:/app
    ports:
      - '8000:8000'
    restart: always
    depends_on:
      - db

  # PostgreSQL
  db:
    image: postgres
    restart: always
    ports:
      - '5432:5432'
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=${POSTGRES_DB}

  db_test:
    image: postgres:14.1-alpine
    restart: always
    environment:
      - POSTGRES_USER=${POSTGRES_USER_TEST}
      - POSTGRES_PASSWORD=${POSTGRES_USER_TEST}
      - POSTGRES_DB=${POSTGRES_DB_TEST}
    ports:
      - '5433:5432'


  # PostgreSQL UI
  pgadmin:
    image: dpage/pgadmin4
    environment:
      - PGADMIN_DEFAULT_EMAIL=${PGADMIN_DEFAULT_EMAIL}
      - PGADMIN_DEFAULT_PASSWORD=${PGADMIN_DEFAULT_PASSWORD}
    ports:
      - '5050:80'
    depends_on:
      - db

env variables:

DATABASE_URL=postgresql+asyncpg://postgres:password@db:5432/marketplace_db

POSTGRES_USER=postgres
POSTGRES_PASSWORD=password
POSTGRES_DB=marketplace_db
POSTGRES_HOST=db
POSTGRES_PORT=5432

POSTGRES_USER_TEST=test_marketplace_user
POSTGRES_PASSWORD_TEST=test_marketplace_password
POSTGRES_DB_TEST=test_marketplace_db
POSTGRES_HOST_TEST=db
POSTGRES_PORT_TEST=5432

[email protected]
PGADMIN_DEFAULT_PASSWORD=password

config file with pass src/config.py:

load_dotenv()

SECRET_KEY = os.environ.get('SECRET_KEY')

# Main DB Settings
DB_USER = os.environ.get('POSTGRES_USER')
DB_PASS = os.environ.get('POSTGRES_PASSWORD')
DB_NAME = os.environ.get('POSTGRES_DB')
DB_HOST = os.environ.get('POSTGRES_HOST')
DB_PORT = os.environ.get('POSTGRES_PORT')
DATABASE_URL = os.environ.get('DATABASE_URL')

# Test DB Settings
DB_USER_TEST = os.environ.get('POSTGRES_USER_TEST')
DB_PASS_TEST = os.environ.get('POSTGRES_PASSWORD_TEST')
DB_NAME_TEST = os.environ.get('POSTGRES_DB_TEST')
DB_HOST_TEST = os.environ.get('POSTGRES_HOST_TEST')
DB_PORT_TEST = os.environ.get('POSTGRES_PORT_TEST')

and conftest file for pytest with path tests/conftest.py

import asyncio
from typing import AsyncGenerator

import pytest
from httpx import AsyncClient
from sqlalchemy import NullPool
from sqlalchemy.ext.asyncio import (AsyncSession, async_sessionmaker,
                                    create_async_engine)
from fastapi.testclient import TestClient

from src.config import (DB_HOST_TEST, DB_PORT_TEST, DB_NAME_TEST,
                        DB_PASS_TEST, DB_USER_TEST)
from src.database import get_async_session, metadata
from src.main import app

DATABASE_URL_TEST = f"postgresql+asyncpg://{DB_USER_TEST}:{DB_PASS_TEST}@{DB_HOST_TEST}:{DB_PORT_TEST}/{DB_NAME_TEST}"

engine_test = create_async_engine(DATABASE_URL_TEST, poolclass=NullPool)
async_session_maker = async_sessionmaker(engine_test, expire_on_commit=False)
metadata.bind = engine_test


async def override_get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


app.dependency_overrides[get_async_session] = override_get_async_session


@pytest.fixture(autouse=True, scope='session')
async def prepare_database():
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.create_all)
    yield
    async with engine_test.begin() as conn:
        await conn.run_sync(metadata.drop_all)


@pytest.fixture(scope='session')
def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


client = TestClient(app)


@pytest.fixture(scope="session")
async def ac() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

and error message I get when running pytest:

asyncpg.exceptions.InvalidPasswordError: password authentication failed for user "test_marketplace_user"

I have been trying to solve the problem for more than two days and have no success. I don't understand the issue. When I run only the db service and connect to the main database with it's settings (not database for tests) it works fine, except the fact that it is not deleted automatically after each test. Or maybe there are other solutions how to use database for tests and main database in my case? I would be very grateful if someone helps..

Answer 1

I think the line POSTGRES_PASSWORD=${POSTGRES_USER_TEST} in your docker-compose.yml under db_test is a typo.