无法从主机连接Docker内测试PostgreSQL运行pytest：连接异常

问题概述 #

开发基于FastAPI、SQLAlchemy 2和psycopg2的应用，使用docker-compose部署两个PostgreSQL容器：一个用于开发，一个用于自动化测试。参考示例应用容器化后基础测试正常，但后续出现以下问题：

• 后端服务访问开发数据库完全正常
• PyCharm中使用Docker解释器运行pytest时，出现密码认证失败的OperationalError
• 本地虚拟环境执行python -m pytest backend时，出现连接意外关闭的OperationalError

已尝试方案：单独运行测试库并映射端口到5432、将DB_URI中的localhost替换为Docker服务名，均未解决。

环境配置 #

完整目录结构 #

├── README.md
├── backend
│   ├── Dockerfile
│   ├── README.md
│   ├── __init__.py
│   ├── alembic.ini
│   ├── api
│   ├── config
│   │   ├── local.env
│   │   └── test.env
│   ├── config.py
│   ├── db.py
│   ├── main.py
│   ├── migrations
│   ├── models
│   ├── requirements.lock
│   ├── requirements.txt
│   ├── tests
│   │   ├── api
│   │   │   ├── __init__.py
│   │   │   └── guests
│   │   │       ├── __init.py__
│   │   │       ├── __pycache__
│   │   │       │   └── test_views.cpython-311-pytest-7.4.0.pyc
│   │   │       └── test_views.py
│   │   ├── conftest.py
│   │   ├── test_main.py
│   │   └── utils.py
│   └── utils
├── db.password.txt
├── docker-compose.yml
├── pyproject.toml

docker-compose.yml #

version: "3.8"
services:
  backend:
    build: ./backend
    environment:
      - POSTGRES_USER=fastapi_traefik
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=fastapi_traefik
      - APP_CONFIG_FILE=local
    command: bash -c 'while !</dev/tcp/db/5432; do sleep 1; done; cd backend; echo "RUNNING ALEMBIC MIGRATIONS"; alembic upgrade head; cd ..; uvicorn backend.main:app --host 0.0.0.0 --port 80 --reload'
    ports:
      - 80:80
    volumes:
      - .:/app
    depends_on:
      db:
        condition: service_healthy
    secrets:
      - db_password
    networks:
      - dock-db-test

  db:
    image: postgres:15-alpine
    container_name: dev-db
    volumes:
      - postgres_data:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=fastapi_traefik
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=fastapi_traefik
    ports:
      - 5432:5432
    restart: unless-stopped
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready -U fastapi_traefik" ]
      interval: 5s
      timeout: 5s
      retries: 5
    secrets:
      - db_password
    networks:
      - dock-db-test

  test_db:
    image: postgres:15-alpine
    container_name: test-db
    environment:
      - POSTGRES_USER=inclasstoday
      - POSTGRES_PASSWORD=inclasstoday1
      - POSTGRES_DB=student-data_test
      - APP_CONFIG_FILE=test
      - POSTGRES_HOST_AUTH_METHOD="trust"
    restart: unless-stopped
    volumes:
      - postgres_data_test:/var/lib/postgresql/data/
    ports:
      - 5433:5433
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready -U inclasstoday -D student-data_test" ]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - dock-db-test
    command: -p 5433


volumes:
  postgres_data:
  postgres_data_test:


secrets:
  db_password:
    file: db.password.txt

networks:
  dock-db-test:
    external: false
    name: dock-db-test

Local.env #

DB_URI=postgresql+asyncpg://fastapi_traefik:password@db:5432/fastapi_traefik
ECHO_SQL=False

Test.env #

DB_URI=postgresql+asyncpg://inclasstoday:inclasstoday1@localhost:5433/student-data_test
ECHO_SQL=False

Conftest.py #

from typing import AsyncGenerator, Generator

import pytest
from httpx import AsyncClient
from sqlalchemy import create_engine, event, text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import Session, SessionTransaction

from backend.db import get_session
from backend.main import app
from backend.models.base import Base
from backend.config import settings


@pytest.fixture
async def ac() -> AsyncGenerator:
    async with AsyncClient(app=app, base_url="https://test") as c:
        yield c


@pytest.fixture(scope="session")
def setup_db() -> Generator:
    engine = create_engine(f"{settings.DB_URI.replace('+asyncpg', '')}")
    conn = engine.connect()
    conn.execute(text("commit"))
    try:
        conn.execute(text("drop database test"))
    except SQLAlchemyError:
        pass
    finally:
        conn.close()

    conn = engine.connect()
    conn.execute(text("commit"))
    conn.execute(text("create database test"))
    conn.close()

    yield

    conn = engine.connect()
    conn.execute(text("commit"))
    try:
        conn.execute(text("drop database test"))
    except SQLAlchemyError:
        pass
    conn.close()


@pytest.fixture(scope="session", autouse=True)
def setup_test_db(setup_db: Generator) -> Generator:
    engine = create_engine(f"{settings.DB_URI.replace('+asyncpg', '')}/test")

    with engine.begin():
        Base.metadata.drop_all(engine)
        Base.metadata.create_all(engine)
        yield
        Base.metadata.drop_all(engine)


@pytest.fixture
async def session() -> AsyncGenerator:
    async_engine = create_async_engine(f"{settings.DB_URI}/test")
    async with async_engine.connect() as conn:
        await conn.begin()
        await conn.begin_nested()
        AsyncSessionLocal = async_sessionmaker(
            autocommit=False,
            autoflush=False,
            bind=conn,
            future=True,
        )

        async_session = AsyncSessionLocal()

        @event.listens_for(async_session.sync_session, "after_transaction_end")
        def end_savepoint(session: Session, transaction: SessionTransaction) -> None:
            if conn.closed:
                return
            if not conn.in_nested_transaction():
                if conn.sync_connection:
                    conn.sync_connection.begin_nested()

        def test_get_session() -> Generator:
            try:
                yield AsyncSessionLocal
            except SQLAlchemyError:
                pass

        app.dependency_overrides[get_session] = test_get_session

        yield async_session
        await async_session.close()
        await conn.rollback()

config.py #

from pydantic_settings import BaseSettings, SettingsConfigDict
import os
from pathlib import Path

class Settings(BaseSettings):
    DB_URI: str
    ECHO_SQL: bool

    model_config = SettingsConfigDict(
        env_file=Path(__file__).parent / f"config/{os.environ['APP_CONFIG_FILE']}.env",
        case_sensitive=True,
    )


settings = Settings.model_validate({})

Dockerfile #

FROM python:3.11-slim-buster

WORKDIR /app

COPY ./requirements.lock ./
RUN pip install --no-cache-dir --upgrade -r requirements.lock
COPY . .

错误信息 #

本地虚拟环境运行测试时的报错：

return self.loaded_dbapi.connect(*cargs, **cparams)
        cargs      = ()
        cparams    = {'dbname': 'student-data_test', 'host': 'localhost', 'password': 'inclasstoday1', 'port': 5433, ...}
        self       = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x110d4bd50>
venv/lib/python3.11/site-packages/psycopg2/__init__.py:122: in connect
    conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
E   sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at "localhost" (::1), port 5433 failed: server closed the connection unexpectedly
E       This probably means the server terminated abnormally
E       before or while processing the request

问题排查与解决方案 #

1. 修复测试数据库端口映射 #

docker-compose中test_db的端口映射配置错误：PostgreSQL容器默认监听内部端口5432，你通过command: -p 5433修改了容器内端口，但端口映射规则是主机端口:容器内部端口，正确配置应为：

# test_db服务下的端口映射
ports:
  - "5433:5432"
# 移除command: -p 5433，保持容器内部默认端口5432

2. 区分环境配置DB_URI #

• 本地虚拟环境：DB_URI使用localhost:5433（对应主机映射端口），当前Test.env配置正确
• Docker解释器环境：代码运行在Docker网络内，需用服务名test_db代替localhost，端口用容器内部端口5432，新增test-docker.env：

  DB_URI=postgresql+asyncpg://inclasstoday:inclasstoday1@test_db:5432/student-data_test
  ECHO_SQL=False
  

  运行测试时设置环境变量APP_CONFIG_FILE=test-docker

3. 修正测试数据库健康检查命令 #

pg_isready没有-D参数，正确指定数据库名的参数是-d，同时指定端口：

healthcheck:
  test: [ "CMD-SHELL", "pg_isready -U inclasstoday -d student-data_test -p 5432" ]

4. 修复PostgreSQL认证配置 #

POSTGRES_HOST_AUTH_METHOD="trust"的引号会导致配置异常，去掉引号：

environment:
  - POSTGRES_HOST_AUTH_METHOD=trust

注意：测试环境可临时使用trust认证，若需密码认证则移除该配置，保持默认md5认证。

5. 调整测试库创建逻辑 #

setup_db fixture中直接连接目标数据库创建测试库会失败，需先连接PostgreSQL默认的postgres库：

@pytest.fixture(scope="session")
def setup_db() -> Generator:
    # 连接postgres默认库执行测试库创建/删除操作
    base_uri = settings.DB_URI.replace('+asyncpg', '').replace('/student-data_test', '/postgres')
    engine = create_engine(base_uri)
    conn = engine.connect()
    conn.execute(text("commit"))
    
    try:
        conn.execute(text("drop database if exists test"))
    except SQLAlchemyError:
        pass
    conn.execute(text("create database test"))
    conn.close()

    yield

    # 清理测试库
    conn = engine.connect()
    conn.execute(text("commit"))
    try:
        conn.execute(text("drop database if exists test"))
    except SQLAlchemyError:
        pass
    conn.close()

内容的提问来源于stack exchange，提问作者Fred Chasin