Pytest 和 PostgreSQL：每次测试的新数据库

def test_create_admin_table(test_db):    ...

def test_create_admin_table(test_db):    # open a cursor to perform database operations    cur = test_db.cursor()    # pass data to fill a query placeholders and let psycopg perform    # the correct conversion (no sql injections!)    cur.execute(        "insert into test (num, data) values (%s, %s)",        (100, "abc'def"))    # query the database and obtain data as python objects.    cur.execute("select * from test")    cur.fetchone()    # will return (1, 100, "abc'def")    # you can use `cur.fetchmany()`, `cur.fetchall()` to return a list    # of several records, or even iterate on the cursor    for record in cur:        print(record)

├── src│   └── tuvok│       ├── __init__.py│       └── sales│           └── new_user.py├── tests│   ├── conftest.py│   └── sales│       └── test_new_user.py├── requirements.txt└── yoyo.ini

├── migrations    ├── 20240816_01_yn3ca-sales-user-user-add-last-run-table.py    ├── ...

# without db name!test_db_url = "postgresql://localhost"test_db_name = "test_tuvok"test_db_migrations_dir = str(path(__file__, "../../migrations").resolve())

@pytest.fixturedef test_db():    # autocommit=true start no transaction because create/drop database    # cannot be executed in a transaction block.    with psycopg.connect(test_db_url, autocommit=true) as conn:        cur = conn.cursor()        # create test db, drop before        cur.execute(f'drop database if exists "{test_db_name}" with (force)')        cur.execute(f'create database "{test_db_name}"')        # return (a new) connection to just created test db        # unfortunately, you cannot directly change the database for an existing psycopg connection. once a connection is established to a specific database, it's tied to that database.        with psycopg.connect(test_db_url, dbname=test_db_name) as conn:            yield conn        cur.execute(f'drop database if exists "{test_db_name}" with (force)')

@pytest.fixturedef yoyo():    # yoyo expect `driver://user:pass@host:port/database_name?param=value`.    # in passed url we need to    url = (        urlparse(test_db_url)        .        # 1) change driver (schema part) with `postgresql+psycopg` to use        # psycopg 3 (not 2 which is `postgresql+psycopg2`)        _replace(scheme="postgresql+psycopg")        .        # 2) change database to test db (in which migrations will apply)        _replace(path=test_db_name)        .geturl()    )    backend = get_backend(url)    migrations = read_migrations(test_db_migrations_dir)    if len(migrations) == 0:        raise valueerror(f"no yoyo migrations found in '{test_db_migrations_dir}'")    with backend.lock():        backend.apply_migrations(backend.to_apply(migrations))

@pytest.fixturedef test_db(yoyo):    ...

def test_create_admin_table(test_db, yoyo):    ...