未找到测试夹具'postgres'

Question

我正在设置unittest来测试气流操作员的postgres连接。我有一个设置函数来启动postgres容器，然后还有一个函数来测试容器中的一些查询。我对此还比较陌生，所以毫无疑问这是我的逻辑错误。

class TestOperator:
def setUp(self):
#pytest postresql container patches
postgres_image = fetch(repository="postgres:11.1-alpine")
postgres = container(
image="{postgres_image.id}",
environment={"POSTGRES_USER": "testuser", "POSTGRES_PASSWORD": "testpass"},
ports={"5432/tcp": None},
volumes={
os.path.join(os.path.dirname(__file__), "postgres-init.sql"): {
"bind": "/docker-entrypoint-initdb.d/postgres-init.sql"
}
}
)

"""
Using Pytest Mocker to create a Postgresql container to connect test.
"""
def test_postgres_operator(self, mocker, postgres):
mocker.patch.object(
PostgresHook,
"get_connection",
return_value=Connection(
conn_id="postgres",
conn_type="postgres",
host="localhost",
login="testuser",
password="testpass",
port=postgres.ports["5432/tcp"][0],
),
)
#target Postgres Container for
task = PostGresOperator(
task_id="PostgresOperator",
postgres_conn_id="postgres_id",
)
pg_hook = PostgresHook()
row_count = pg_hook.get_first("select * from test")[0]
assert row_count >1

然后我得到错误

fixture 'postgres' not found

我确信我的逻辑是错误的。

Answer 1

pytest认为postgres是一个固定装置，但找不到它。

您可以将is设置为实例字段:，而不是将postgres作为参数传入

def setUp(self):
postgres_image = fetch(repository="postgres:11.1-alpine")
self.postgres = container(...)
def test_postgres_operator(self, mocker):
# use self.postgres instead of postgres

或者，您可以将postgres定义为适当的pytest固定装置，以提高可重用性。

或者，您可以查看pytest-postgresql插件，这可能会使模拟和测试postgresql相关代码变得更容易。