Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

mock_rds - postgresql+psycopg2 #81

Open
ayshamdmgit opened this issue Jan 7, 2020 · 0 comments
Open

mock_rds - postgresql+psycopg2 #81

ayshamdmgit opened this issue Jan 7, 2020 · 0 comments

Comments

@ayshamdmgit
Copy link

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant