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

Fix:Add list connection API's (#3976)

Browse Source 
* Initial commit create connection

* Added connection API's

* Add destination API's

* fixed tests

* fix tests

* Fixed tests
This commit is contained in:
usmangani1 2021-06-07 15:05:28 +05:30 committed by GitHub
parent 74559f2a91
commit d325593e46
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 253 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
moto/events/models.py
View File

@ -542,6 +542,51 @@ class Replay(BaseModel):
self.end_time = unix_time(datetime.utcnow())
class Connection(BaseModel):
def __init__(
self, name, region_name, description, authorization_type, auth_parameters,
):
self.name = name
self.region = region_name
self.description = description
self.authorization_type = authorization_type
self.auth_parameters = auth_parameters
self.creation_time = unix_time(datetime.utcnow())
self.state = "AUTHORIZED"
@property
def arn(self):
return "arn:aws:events:{0}:{1}:connection/{2}".format(
self.region, ACCOUNT_ID, self.name
)
class Destination(BaseModel):
def __init__(
self,
name,
region_name,
description,
connection_arn,
invocation_endpoint,
http_method,
):
self.name = name
self.region = region_name
self.description = description
self.connection_arn = connection_arn
self.invocation_endpoint = invocation_endpoint
self.creation_time = unix_time(datetime.utcnow())
self.http_method = http_method
self.state = "ACTIVE"
@property
def arn(self):
return "arn:aws:events:{0}:{1}:destination/{2}".format(
self.region, ACCOUNT_ID, self.name
)
class EventPattern:
def __init__(self, filter):
self._filter = self._load_event_pattern(filter)
@ -649,6 +694,8 @@ class EventsBackend(BaseBackend):
self.tagger = TaggingService()
self._add_default_event_bus()
self.connections = {}
self.destinations = {}
def reset(self):
region_name = self.region_name
@ -803,7 +850,6 @@ class EventsBackend(BaseBackend):
raise ValidationException(
"ScheduleExpression is supported only on the default event bus."
)
if name in self.rules:
self.update_rule(self.rules[name], **kwargs)
new_rule = self.rules[name]
@ -1282,6 +1328,38 @@ class EventsBackend(BaseBackend):
return {"ReplayArn": replay.arn, "State": ReplayState.CANCELLING.value}
def create_connection(self, name, description, authorization_type, auth_parameters):
connection = Connection(
name, self.region_name, description, authorization_type, auth_parameters
)
self.connections[name] = connection
return connection
def list_connections(self):
return self.connections.values()
def create_api_destination(
self, name, description, connection_arn, invocation_endpoint, http_method
):
destination = Destination(
name=name,
region_name=self.region_name,
description=description,
connection_arn=connection_arn,
invocation_endpoint=invocation_endpoint,
http_method=http_method,
)
self.destinations[name] = destination
return destination
def list_api_destinations(self):
return self.destinations.values()
def describe_api_destination(self, name):
return self.destinations.get(name)
events_backends = {}
for region in Session().get_available_regions("events"):

99
moto/events/responses.py
View File

@ -409,3 +409,102 @@ class EventsHandler(BaseResponse):
result = self.events_backend.cancel_replay(name)
return json.dumps(result), self.response_headers
def create_connection(self):
name = self._get_param("Name")
description = self._get_param("Description")
authorization_type = self._get_param("AuthorizationType")
auth_parameters = self._get_param("AuthParameters")
result = self.events_backend.create_connection(
name, description, authorization_type, auth_parameters
)
return (
json.dumps(
{
"ConnectionArn": result.arn,
"ConnectionState": "AUTHORIZED",
"CreationTime": result.creation_time,
"LastModifiedTime": result.creation_time,
}
),
self.response_headers,
)
def list_connections(self):
connections = self.events_backend.list_connections()
result = []
for connection in connections:
result.append(
{
"ConnectionArn": connection.arn,
"ConnectionState": "AUTHORIZED",
"CreationTime": connection.creation_time,
"LastModifiedTime": connection.creation_time,
"AuthorizationType": connection.authorization_type,
}
)
return json.dumps({"Connections": result}), self.response_headers
def create_api_destination(self):
name = self._get_param("Name")
description = self._get_param("Description")
connection_arn = self._get_param("ConnectionArn")
invocation_endpoint = self._get_param("InvocationEndpoint")
http_method = self._get_param("HttpMethod")
destination = self.events_backend.create_api_destination(
name, description, connection_arn, invocation_endpoint, http_method
)
return (
json.dumps(
{
"ApiDestinationArn": destination.arn,
"ApiDestinationState": "ACTIVE",
"CreationTime": destination.creation_time,
"LastModifiedTime": destination.creation_time,
}
),
self.response_headers,
)
def list_api_destinations(self):
destinations = self.events_backend.list_api_destinations()
result = []
for destination in destinations:
result.append(
{
"ApiDestinationArn": destination.arn,
"Name": destination.name,
"ApiDestinationState": destination.state,
"ConnectionArn": destination.connection_arn,
"InvocationEndpoint": destination.invocation_endpoint,
"HttpMethod": destination.http_method,
"CreationTime": destination.creation_time,
"LastModifiedTime": destination.creation_time,
}
)
return json.dumps({"ApiDestinations": result}), self.response_headers
def describe_api_destination(self):
name = self._get_param("Name")
destination = self.events_backend.describe_api_destination(name)
return (
json.dumps(
{
"ApiDestinationArn": destination.arn,
"Name": destination.name,
"ApiDestinationState": destination.state,
"ConnectionArn": destination.connection_arn,
"InvocationEndpoint": destination.invocation_endpoint,
"HttpMethod": destination.http_method,
"CreationTime": destination.creation_time,
"LastModifiedTime": destination.creation_time,
}
),
self.response_headers,
)

75
tests/test_events/test_events.py
View File

@ -2138,3 +2138,78 @@ def test_start_replay_send_to_log_group():
event_replay["resources"].should.be.empty
event_replay["detail"].should.equal({"key": "value"})
event_replay["replay-name"].should.equal("test-replay")
@mock_events
def test_create_and_list_connections():
client = boto3.client("events", "eu-central-1")
response = client.list_connections()
assert len(response.get("Connections")) == 0
response = client.create_connection(
Name="test",
Description="test description",
AuthorizationType="API_KEY",
AuthParameters={
"ApiKeyAuthParameters": {"ApiKeyName": "test", "ApiKeyValue": "test"}
},
)
assert response.get(
"ConnectionArn"
) == "arn:aws:events:eu-central-1:{0}:connection/test".format(ACCOUNT_ID)
response = client.list_connections()
assert response.get("Connections")[0].get(
"ConnectionArn"
) == "arn:aws:events:eu-central-1:{0}:connection/test".format(ACCOUNT_ID)
@mock_events
def test_create_and_list_api_destinations():
client = boto3.client("events", "eu-central-1")
response = client.create_connection(
Name="test",
Description="test description",
AuthorizationType="API_KEY",
AuthParameters={
"ApiKeyAuthParameters": {"ApiKeyName": "test", "ApiKeyValue": "test"}
},
)
destination_response = client.create_api_destination(
Name="test",
Description="test-description",
ConnectionArn=response.get("ConnectionArn"),
InvocationEndpoint="www.google.com",
HttpMethod="GET",
)
assert destination_response.get(
"ApiDestinationArn"
) == "arn:aws:events:eu-central-1:{0}:destination/test".format(ACCOUNT_ID)
assert destination_response.get("ApiDestinationState") == "ACTIVE"
destination_response = client.describe_api_destination(Name="test")
assert destination_response.get(
"ApiDestinationArn"
) == "arn:aws:events:eu-central-1:{0}:destination/test".format(ACCOUNT_ID)
assert destination_response.get("Name") == "test"
assert destination_response.get("ApiDestinationState") == "ACTIVE"
destination_response = client.list_api_destinations()
assert destination_response.get("ApiDestinations")[0].get(
"ApiDestinationArn"
) == "arn:aws:events:eu-central-1:{0}:destination/test".format(ACCOUNT_ID)
assert destination_response.get("ApiDestinations")[0].get("Name") == "test"
assert (
destination_response.get("ApiDestinations")[0].get("ApiDestinationState")
== "ACTIVE"
)