I am trying to mock the rds resource using moto and have done it in the below manner
from moto import mock_rds2
import boto3
from sqlalchemy import create_engine
import pytest
@pytest.fixture(scope='function')
def aws_credentials():
"""Mocked AWS Credentials for moto."""
os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
os.environ['AWS_SECURITY_TOKEN'] = 'testing'
os.environ['AWS_SESSION_TOKEN'] = 'testing'
@pytest.fixture(scope='function', autouse=True)
def rds(aws_credentials):
with mock_rds2():
yield boto3.client('rds', region_name='us-east-1')
@pytest.fixture(scope='function', autouse=True)
def rds_instance(rds):
"""RDS mock service"""
test_instances = rds.create_db_instance(
DBName='test_db',
AllocatedStorage=10,
StorageType='standard',
DBInstanceIdentifier='instance',
DBInstanceClass='db.t2.micro',
Engine='postgres',
MasterUsername='postgres_user',
MasterUserPassword='p$ssw$rd',
AvailabilityZone='us-east-1',
PubliclyAccessible=True,
DBSecurityGroups=["my_sg"],
VpcSecurityGroupIds=["sg-123456"]
)
yield test_instances
# print(rds.describe_db_instances())
#
# db_instance = test_instances["DBInstance"]
# print(db_instance)
# user_name = db_instance['MasterUsername']
# host = db_instance['Endpoint']['Address']
# port = db_instance['Endpoint']['Port']
# db_name = db_instance['DBName']
# conn_str = f'postgresql+psycopg2://{user_name}:p$ssw$rd@{host}:{port}/{db_name}'
# print(conn_str)
# engine_con = create_engine(conn_str)
# engine_con.connect()
def test_conn(rds_instance):
db_instance = rds_instance["DBInstance"]
print(db_instance)
user_name = db_instance['MasterUsername']
host = db_instance['Endpoint']['Address']
port = db_instance['Endpoint']['Port']
db_name = db_instance['DBName']
conn_str = f'postgresql+psycopg2://{user_name}:p$ssw$rd@{host}:{port}/{db_name}'
print(conn_str)
engine_con = create_engine(conn_str)
engine_con.connect()
I am getting an error at the "engine_con.connect()"
sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "instance.aaaaaaaaaa.us-east-1.rds.amazonaws.com" to address: Unknown host
E
I am trying to mock the rds resource using moto and have done it in the below manner
from moto import mock_rds2
import boto3
from sqlalchemy import create_engine
import pytest
@pytest.fixture(scope='function')
def aws_credentials():
"""Mocked AWS Credentials for moto."""
os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
os.environ['AWS_SECURITY_TOKEN'] = 'testing'
os.environ['AWS_SESSION_TOKEN'] = 'testing'
@pytest.fixture(scope='function', autouse=True)
def rds(aws_credentials):
with mock_rds2():
yield boto3.client('rds', region_name='us-east-1')
@pytest.fixture(scope='function', autouse=True)
def rds_instance(rds):
"""RDS mock service"""
test_instances = rds.create_db_instance(
DBName='test_db',
AllocatedStorage=10,
StorageType='standard',
DBInstanceIdentifier='instance',
DBInstanceClass='db.t2.micro',
Engine='postgres',
MasterUsername='postgres_user',
MasterUserPassword='p$ssw$rd',
AvailabilityZone='us-east-1',
PubliclyAccessible=True,
DBSecurityGroups=["my_sg"],
VpcSecurityGroupIds=["sg-123456"]
)
yield test_instances
# print(rds.describe_db_instances())
#
# db_instance = test_instances["DBInstance"]
# print(db_instance)
# user_name = db_instance['MasterUsername']
# host = db_instance['Endpoint']['Address']
# port = db_instance['Endpoint']['Port']
# db_name = db_instance['DBName']
# conn_str = f'postgresql+psycopg2://{user_name}:p$ssw$rd@{host}:{port}/{db_name}'
# print(conn_str)
# engine_con = create_engine(conn_str)
# engine_con.connect()
def test_conn(rds_instance):
db_instance = rds_instance["DBInstance"]
print(db_instance)
user_name = db_instance['MasterUsername']
host = db_instance['Endpoint']['Address']
port = db_instance['Endpoint']['Port']
db_name = db_instance['DBName']
conn_str = f'postgresql+psycopg2://{user_name}:p$ssw$rd@{host}:{port}/{db_name}'
print(conn_str)
engine_con = create_engine(conn_str)
engine_con.connect()
I am getting an error at the "engine_con.connect()"
sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "instance.aaaaaaaaaa.us-east-1.rds.amazonaws.com" to address: Unknown host
E