尝试在 python 测试中模拟红移时出错

Question

我正在尝试使用 @mock_redshift 的 moto 来模拟与 AWS Redshift 的连接，我正在使用 boto3 创建测试集群，但是当我进行查询时，我收到以下错误：

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

看看Cluster 对moto 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) 语法。