Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix up some issues found in unittests - comment out snapshotting until later

Browse Source
This commit is contained in:
rocky4570fft 2017-08-14 00:27:15 +10:00
parent 9687b6e03e
commit ed39c2ea4e
3 changed files with 129 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
moto/rds2/exceptions.py
View File

@ -58,3 +58,18 @@ class DBParameterGroupNotFoundError(RDSClientError):
super(DBParameterGroupNotFoundError, self).__init__(
'DBParameterGroupNotFound',
'DB Parameter Group {0} not found.'.format(db_parameter_group_name))
class InvalidDBClusterStateFaultError(RDSClientError):
def __init__(self, database_identifier):
super(InvalidDBClusterStateFaultError, self).__init__(
'InvalidDBClusterStateFault',
'Invalid DB type, when trying to perform StopDBInstance on {0}e. See AWS RDS documentation on rds.stop_db_instance'.format(database_identifier))
class InvalidDBInstanceStateError(RDSClientError):
def __init__(self, database_identifier, istate):
estate = "in available state" if istate == 'stop' else "stopped, it cannot be started"
super(InvalidDBInstanceStateError, self).__init__(
'InvalidDBInstanceState',
'when calling the {}DBInstance operation: Instance {} is not {}.'.format(istate.title(), database_identifier, estate))

34
moto/rds2/models.py
View File

@ -18,7 +18,9 @@ from .exceptions import (RDSClientError,
DBSnapshotNotFoundError,
DBSecurityGroupNotFoundError,
DBSubnetGroupNotFoundError,
DBParameterGroupNotFoundError)
DBParameterGroupNotFoundError,
InvalidDBClusterStateFaultError,
InvalidDBInstanceStateError)
class Database(BaseModel):
@ -737,28 +739,30 @@ class RDS2Backend(BaseBackend):
database = self.describe_databases(db_instance_identifier)[0]
# todo: certain rds types not allowed to be stopped at this time.
if database.is_replica or database.multi_az:
# should be 400 error
return RDSClientError('InvalidDBClusterStateFault', 'Invalid DB type, when trying to perform StopDBInstance. See AWS RDS documentation on rds.stop_db_instance')
# todo: more db types not supported by stop/start instance api
raise InvalidDBClusterStateFaultError(db_instance_identifier)
if database.status != 'available':
return RDSClientError('InvalidDBInstanceState', 'when calling the StopDBInstance operation: Instance %s is not in available state' % db_instance_identifier)
if db_snapshot_identifier:
self.create_rds_snapshot(db_instance_identifier, db_snapshot_identifier)
raise InvalidDBInstanceStateError(db_instance_identifier, 'stop')
# todo: create rds snapshots
# if db_snapshot_identifier:
# self.create_rds_snapshot(db_instance_identifier, db_snapshot_identifier)
database.status = 'shutdown'
return database
def start_database(self, db_instance_identifier):
database = self.describe_databases(db_instance_identifier)[0]
if database.status != 'shutdown': # should be 400 error
return RDSClientError('InvalidDBInstanceState', 'when calling the StartDBInstance operation: Instance %s is not stopped, it cannot be started.' % db_instance_identifier)
# todo: bunch of different error messages to be generated from this api call
if database.status != 'shutdown':
raise InvalidDBInstanceStateError(db_instance_identifier, 'start')
database.status = 'available'
return
return database
def create_rds_snapshot(self, db_instance_identifier, db_snapshot_identifier):
database = self.describe_databases(db_instance_identifier)[0]
# todo
# DBSnapshotAlreadyExists
# SnapshotQuotaExceeded
return None
# def create_rds_snapshot(self, db_instance_identifier, db_snapshot_identifier):
# database = self.describe_databases(db_instance_identifier)[0]
# # todo
# # DBSnapshotAlreadyExists
# # SnapshotQuotaExceeded
# return None
def find_db_from_id(self, db_id):
if self.arn_regex.match(db_id):

98
tests/test_rds2/test_rds2.py
View File

@ -19,9 +19,6 @@ def test_create_database():
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
database['DBInstance']['DBInstanceStatus'].should.equal('available')
database['DBInstance']['DBName'].should.equal('staging-postgres')
database['DBInstance']['DBInstanceIdentifier'].should.equal("db-master-1")
database['DBInstance']['AllocatedStorage'].should.equal(10)
database['DBInstance']['DBInstanceClass'].should.equal("db.m1.small")
database['DBInstance']['LicenseModel'].should.equal("license-included")
@ -30,7 +27,102 @@ def test_create_database():
'DBSecurityGroupName'].should.equal('my_sg')
database['DBInstance']['DBInstanceArn'].should.equal(
'arn:aws:rds:us-west-2:1234567890:db:db-master-1')
database['DBInstance']['DBInstanceStatus'].should.equal('available')
database['DBInstance']['DBName'].should.equal('staging-postgres')
database['DBInstance']['DBInstanceIdentifier'].should.equal("db-master-1")
@mock_rds2
def test_stop_database():
conn = boto3.client('rds', region_name='us-west-2')
database = conn.create_db_instance(DBInstanceIdentifier='db-master-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
LicenseModel='license-included',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
mydb = conn.describe_db_instances(DBInstanceIdentifier=database['DBInstance']['DBInstanceIdentifier'])['DBInstances'][0]
mydb['DBInstanceStatus'].should.equal('available')
# test stopping database
response = conn.stop_db_instance(DBInstanceIdentifier=mydb['DBInstanceIdentifier'])
response['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
response['DBInstance']['DBInstanceStatus'].should.equal('shutdown')
# test rdsclient error when trying to stop an already stopped database
conn.stop_db_instance.when.called_with(DBInstanceIdentifier=mydb['DBInstanceIdentifier']).should.throw(ClientError)
@mock_rds2
def test_start_database():
conn = boto3.client('rds', region_name='us-west-2')
database = conn.create_db_instance(DBInstanceIdentifier='db-master-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
LicenseModel='license-included',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
mydb = conn.describe_db_instances(DBInstanceIdentifier=database['DBInstance']['DBInstanceIdentifier'])['DBInstances'][0]
mydb['DBInstanceStatus'].should.equal('available')
# test trying to start an already started database
conn.start_db_instance.when.called_with(DBInstanceIdentifier=mydb['DBInstanceIdentifier']).should.throw(ClientError)
# stop and test start - should go from shutdown to available
response = conn.stop_db_instance(DBInstanceIdentifier=mydb['DBInstanceIdentifier'])
response['DBInstance']['DBInstanceStatus'].should.equal('shutdown')
response = conn.start_db_instance(DBInstanceIdentifier=mydb['DBInstanceIdentifier'])
response['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
response['DBInstance']['DBInstanceStatus'].should.equal('available')
@mock_rds2
def test_fail_to_stop_multi_az():
conn = boto3.client('rds', region_name='us-west-2')
database = conn.create_db_instance(DBInstanceIdentifier='db-master-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
LicenseModel='license-included',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"],
MultiAZ=True)
mydb = conn.describe_db_instances(DBInstanceIdentifier=database['DBInstance']['DBInstanceIdentifier'])['DBInstances'][0]
mydb['DBInstanceStatus'].should.equal('available')
# multi-az databases arent allowed to be shutdown at this time.
conn.stop_db_instance.when.called_with(DBInstanceIdentifier=mydb['DBInstanceIdentifier']).should.throw(ClientError)
# multi-az databases arent allowed to be started up at this time.
conn.start_db_instance.when.called_with(DBInstanceIdentifier=mydb['DBInstanceIdentifier']).should.throw(ClientError)
@mock_rds2
def test_fail_to_stop_readreplica():
conn = boto3.client('rds', region_name='us-west-2')
database = conn.create_db_instance(DBInstanceIdentifier='db-master-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
LicenseModel='license-included',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
replica = conn.create_db_instance_read_replica(DBInstanceIdentifier="db-replica-1",
SourceDBInstanceIdentifier="db-master-1",
DBInstanceClass="db.m1.small")
mydb = conn.describe_db_instances(DBInstanceIdentifier=replica['DBInstance']['DBInstanceIdentifier'])['DBInstances'][0]
mydb['DBInstanceStatus'].should.equal('available')
# read-replicas are not allowed to be stopped at this time.
conn.stop_db_instance.when.called_with(DBInstanceIdentifier=mydb['DBInstanceIdentifier']).should.throw(ClientError)
# read-replicas are not allowed to be started at this time.
conn.start_db_instance.when.called_with(DBInstanceIdentifier=mydb['DBInstanceIdentifier']).should.throw(ClientError)
@mock_rds2
def test_get_databases():