Блог

Я пытаюсь использовать @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()
  

Взгляните на Cluster реализацию moto в их Github.

mock_redshift Функциональность имитирует API управления кластером Redshift от boto3, но не саму базу данных. Соединение с URL, предоставляемым moto (т. Е. test.cg034hpkmmjt.us-east-1.redshift.amazonaws.com ) никогда не будет работать.

Если вы хотите протестировать реальный доступ к Redshift, я думаю, хорошей идеей будет протестировать с контейнером под управлением Postgres, хорошей библиотекой Python, которая легко предоставляет эту функциональность, является testcontainers (ссылка на Github)

Вы можете установить на свой виртуальный env с помощью

 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, на самом деле есть много важных отличий (как указано в этой статье). Например, недавно меня затронуло то, что Redshift не принимает DISTINCT ON (column) синтаксис.