python - 尝试在 python 测试中模拟 redshift 时出错

Question

我正在尝试使用 moto 的 @mock_redshift为了模拟与 AWS Redshift 的连接，我使用 boto3 创建 de test 集群，但是当我进行查询时，出现以下错误:

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "recon-test.cg034hpkmmjt.us-east-1.redshift.amazonaws.com" to address: Name or service not known

代码如下:

@mock_redshift
@mock_s3
@pytest.mark.vcr
def test_extract_stp_both_dates():
    client = boto3.client('redshift', region_name='us-east-1')
    response = client.create_cluster(
        DBName='recon',
        ClusterIdentifier='recon-test',
        ClusterType='single-node',
        NodeType='ds2.xlarge',
        MasterUsername='cuenca',
        MasterUserPassword='password',
    )

    host = response['Cluster']['Endpoint']['Address']
    port = response['Cluster']['Endpoint']['Port']

    rs_client = RedshiftClient(
        'recon',
        'cuenca',
        'password',
        host,
        port,
    )

    rs_client.s.execute("CREATE TABLE table_test (attr VARCHAR);") # The error is here

    conn = boto3.resource('s3', region_name='us-east-1')
    conn.create_bucket(Bucket=os.environ['S3_BUCKET'])
    random.seed(1)
    responses.add_passthru('https://')
    extract('01/11/2018', '30/11/2018', rs_client=rs_client)

这是 RedshiftClient

import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker


class RedshiftClient:
    def __init__(
        self, database: str, user: str, password: str, host: str, port: str
    ):
        self.connection_string = (
            f'redshift+psycopg2://{user}:{password}@{host}:{port}/{database}'
        )
        self.engine = sa.create_engine(self.connection_string)
        self.sessionmaker = sessionmaker(bind=self.engine)
        self.s = self.sessionmaker()

    def exec_query(self, query: str) -> list:
        return self.s.execute(query).fetchall()

Answer 1

看一下 moto 的 Cluster 实现 in their Github .

mock_redshift 功能模拟 boto3 的 Redshift 集群管理 API，而不是数据库本身。与 moto 提供的 URL 的连接(即 test.cg034hpkmmjt.us-east-1.redshift.amazonaws.com)将永远无法工作。

如果您想测试真正的 Redshift 访问，我认为一个好主意是使用运行 Postgres 的容器进行测试，一个可以轻松提供此功能的良好 Python 库是 testcontainers ( Github link )

你可以安装到你的虚拟环境中使用

pip install testcontainers[postgresql]

并修改你的测试以包含

from testcontainers.postgres import PostgresContainer


def test_docker_run_postgress():
    postgres_container = PostgresContainer("postgres:9.5")
    with postgres_container as postgres:
        e = sqlalchemy.create_engine(postgres.get_connection_url())
        result = e.execute("SELECT version()")

我想指出 Postgres 与 Redshift 不同，事实上有许多重要的区别 (like highlighted in this article)。例如，最近影响我的是 Redshift 不接受 DISTINCT ON (column) 语法。