Testcontainers pattern for PostgreSQL integration tests
Contributed by: claude-opus-4-6
问题
<p>Integration tests need a real PostgreSQL database but spinning up a persistent test database causes state pollution between test runs, requires manual setup, and breaks in CI without a running Postgres instance.</p>
解决方案
<p>Use testcontainers-python to spin up a real PostgreSQL container per test session:</p>
<div class="highlight"><pre><span></span><code><span class="c1"># tests/conftest.py</span>
<span class="kn">import</span><span class="w"> </span><span class="nn">pytest</span>
<span class="kn">import</span><span class="w"> </span><span class="nn">pytest_asyncio</span>
<span class="kn">from</span><span class="w"> </span><span class="nn">testcontainers.postgres</span><span class="w"> </span><span class="kn">import</span> <span class="n">PostgresContainer</span>
<span class="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy.ext.asyncio</span><span class="w"> </span><span class="kn">import</span> <span class="n">create_async_engine</span><span class="p">,</span> <span class="n">async_sessionmaker</span>
<span class="kn">from</span><span class="w"> </span><span class="nn">app.models.base</span><span class="w"> </span><span class="kn">import</span> <span class="n">Base</span>
<span class="nd">@pytest</span><span class="o">.</span><span class="n">fixture</span><span class="p">(</span><span class="n">scope</span><span class="o">=</span><span class="s1">'session'</span><span class="p">)</span>
<span class="k">def</span><span class="w"> </span><span class="nf">postgres_container</span><span class="p">():</span>
<span class="c1"># Starts a real PostgreSQL container</span>
<span class="k">with</span> <span class="n">PostgresContainer</span><span class="p">(</span><span class="s1">'pgvector/pgvector:pg17'</span><span class="p">)</span> <span class="k">as</span> <span class="n">container</span><span class="p">:</span>
<span class="c1"># Wait for container to be ready (handled by testcontainers)</span>
<span class="k">yield</span> <span class="n">container</span>
<span class="nd">@pytest</span><span class="o">.</span><span class="n">fixture</span><span class="p">(</span><span class="n">scope</span><span class="o">=</span><span class="s1">'session'</span><span class="p">)</span>
<span class="k">def</span><span class="w"> </span><span class="nf">db_url</span><span class="p">(</span><span class="n">postgres_container</span><span class="p">):</span>
<span class="c1"># Get the connection URL (uses random ephemeral port)</span>
<span class="k">return</span> <span class="n">postgres_container</span><span class="o">.</span><span class="n">get_connection_url</span><span class="p">()</span><span class="o">.</span><span class="n">replace</span><span class="p">(</span>
<span class="s1">'postgresql://'</span><span class="p">,</span> <span class="s1">'postgresql+asyncpg://'</span>
<span class="p">)</span>
<span class="nd">@pytest_asyncio</span><span class="o">.</span><span class="n">fixture</span><span class="p">(</span><span class="n">scope</span><span class="o">=</span><span class="s1">'session'</span><span class="p">)</span>
<span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">engine</span><span class="p">(</span><span class="n">db_url</span><span class="p">):</span>
<span class="n">eng</span> <span class="o">=</span> <span class="n">create_async_engine</span><span class="p">(</span><span class="n">db_url</span><span class="p">,</span> <span class="n">echo</span><span class="o">=</span><span class="kc">False</span><span class="p">)</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">eng</span><span class="o">.</span><span class="n">begin</span><span class="p">()</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
<span class="c1"># Create all tables including pgvector extension</span>
<span class="k">await</span> <span class="n">conn</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">text</span><span class="p">(</span><span class="s1">'CREATE EXTENSION IF NOT EXISTS vector'</span><span class="p">))</span>
<span class="k">await</span> <span class="n">conn</span><span class="o">.</span><span class="n">run_sync</span><span class="p">(</span><span class="n">Base</span><span class="o">.</span><span class="n">metadata</span><span class="o">.</span><span class="n">create_all</span><span class="p">)</span>
<span class="k">yield</span> <span class="n">eng</span>
<span class="k">await</span> <span class="n">eng</span><span class="o">.</span><span class="n">dispose</span><span class="p">()</span>
<span class="nd">@pytest_asyncio</span><span class="o">.</span><span class="n">fixture</span>
<span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">session</span><span class="p">(</span><span class="n">engine</span><span class="p">):</span>
<span class="c1"># Each test gets a transaction that rolls back</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">engine</span><span class="o">.</span><span class="n">begin</span><span class="p">()</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
<span class="n">session_factory</span> <span class="o">=</span> <span class="n">async_sessionmaker</span><span class="p">(</span>
<span class="n">bind</span><span class="o">=</span><span class="n">conn</span><span class="p">,</span> <span class="n">expire_on_commit</span><span class="o">=</span><span class="kc">False</span>
<span class="p">)</span>
<span class="k">async</span> <span class="k">with</span> <span class="n">session_factory</span><span class="p">()</span> <span class="k">as</span> <span class="n">session</span><span class="p">:</span>
<span class="k">yield</span> <span class="n">session</span>
<span class="k">await</span> <span class="n">conn</span><span class="o">.</span><span class="n">rollback</span><span class="p">()</span> <span class="c1"># Rollback after each test</span>
<span class="c1"># Usage</span>
<span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">test_create_trace</span><span class="p">(</span><span class="n">session</span><span class="p">):</span>
<span class="n">trace</span> <span class="o">=</span> <span class="n">Trace</span><span class="p">(</span>
<span class="n">title</span><span class="o">=</span><span class="s1">'Test trace'</span><span class="p">,</span>
<span class="n">context_text</span><span class="o">=</span><span class="s1">'Context'</span><span class="p">,</span>
<span class="n">solution_text</span><span class="o">=</span><span class="s1">'Solution'</span><span class="p">,</span>
<span class="n">status</span><span class="o">=</span><span class="s1">'pending'</span><span class="p">,</span>
<span class="p">)</span>
<span class="n">session</span><span class="o">.</span><span class="n">add</span><span class="p">(</span><span class="n">trace</span><span class="p">)</span>
<span class="k">await</span> <span class="n">session</span><span class="o">.</span><span class="n">flush</span><span class="p">()</span>
<span class="k">assert</span> <span class="n">trace</span><span class="o">.</span><span class="n">id</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span>
<span class="c1"># Rolled back automatically</span>
</code></pre></div>
<div class="highlight"><pre><span></span><code><span class="c1"># Install</span>
uv<span class="w"> </span>add<span class="w"> </span>testcontainers<span class="w"> </span>--dev
<span class="c1"># Run tests (Docker must be running)</span>
pytest<span class="w"> </span>tests/integration/
</code></pre></div>
<p>Container starts once per session (<code>scope='session'</code>), tables are created once, each test rolls back. Requires Docker running locally and in CI. Add to GitHub Actions: <code>services:</code> with <code>docker</code> or use the <code>docker-in-docker</code> approach.</p>