198 lines
5.6 KiB
Python
198 lines
5.6 KiB
Python
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import json
|
|
|
|
import requests
|
|
import yaml
|
|
|
|
from monascaclient import ksclient
|
|
|
|
# events_url = "http://127.0.0.1:8072"
|
|
events_url = "http://192.168.10.4:8072"
|
|
|
|
|
|
def token():
|
|
keystone = {
|
|
'username': 'mini-mon',
|
|
'password': 'password',
|
|
'project': 'test',
|
|
'auth_url': 'http://192.168.10.5:35357/v3'
|
|
}
|
|
ks_client = ksclient.KSClient(**keystone)
|
|
return ks_client.token
|
|
|
|
headers = {
|
|
'X-Auth-User': 'mini-mon',
|
|
'X-Auth-Key': 'password',
|
|
'X-Auth-Token': token(),
|
|
'Accept': 'application/json',
|
|
'User-Agent': 'python-monascaclient',
|
|
'Content-Type': 'application/json'}
|
|
|
|
|
|
def test_events_get():
|
|
|
|
body = {}
|
|
|
|
response = requests.get(url=events_url + "/v2.0/events",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
|
|
json_data = json.loads(response.text)
|
|
|
|
event_id = json_data[0]['id']
|
|
|
|
assert response.status_code == 200
|
|
|
|
response = requests.get(
|
|
url=events_url + "/v2.0/events/{}".format(event_id),
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
|
|
json_data = json.loads(response.text)
|
|
|
|
new_event_id = json_data[0]['id']
|
|
|
|
assert response.status_code == 200
|
|
assert event_id == new_event_id
|
|
print("GET /events success")
|
|
|
|
|
|
def test_events_get_all():
|
|
print("Test GET /events")
|
|
body = {}
|
|
|
|
response = requests.get(url=events_url + "/v2.0/events",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
print("GET /events success")
|
|
|
|
|
|
def test_stream_definition_post():
|
|
print("Test POST /stream-definitions")
|
|
body = {}
|
|
|
|
notif_resp = requests.get(
|
|
url="http://192.168.10.4:8070/v2.0/notification-methods",
|
|
data=json.dumps(body), headers=headers)
|
|
notif_dict = json.loads(notif_resp.text)
|
|
action_id = str(notif_dict['elements'][0]['id'])
|
|
|
|
body = {"fire_criteria": [{"event_type": "compute.instance.create.start"},
|
|
{"event_type": "compute.instance.create.end"}],
|
|
"description": "provisioning duration",
|
|
"name": "func_test_stream_def",
|
|
"group_by": ["instance_id"],
|
|
"expiration": 3000,
|
|
"select": [{"traits": {"tenant_id": "406904"},
|
|
"event_type": "compute.instance.create.*"}],
|
|
"fire_actions": [action_id],
|
|
"expire_actions": [action_id]}
|
|
|
|
response = requests.post(
|
|
url=events_url + "/v2.0/stream-definitions",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
|
|
assert response.status_code == 201
|
|
print("POST /stream-definitions success")
|
|
|
|
|
|
def test_stream_definition_get():
|
|
print("Test GET /stream-definitions")
|
|
body = {}
|
|
|
|
response = requests.get(
|
|
url=events_url + "/v2.0/stream-definitions/",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
assert response.status_code == 200
|
|
print("GET /stream-definitions success")
|
|
|
|
|
|
def test_stream_definition_delete():
|
|
print("Test DELETE /stream-definitions")
|
|
body = {}
|
|
|
|
stream_resp = requests.get(
|
|
url=events_url + "/v2.0/stream-definitions/",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
stream_dict = json.loads(stream_resp.text)
|
|
stream_id = str(stream_dict['elements'][0]['id'])
|
|
response = requests.delete(
|
|
url=events_url + "/v2.0/stream-definitions/{}".format(
|
|
stream_id),
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
assert response.status_code == 204
|
|
print("DELETE /stream-definitions success")
|
|
|
|
|
|
def test_transforms():
|
|
|
|
print("Test POST /transforms")
|
|
|
|
# Open example yaml file and post to DB
|
|
fh = open('transform_definitions.yaml', 'r')
|
|
specification_data = yaml.load(fh)
|
|
|
|
body = {
|
|
"name": 'func test',
|
|
"description": 'an example definition',
|
|
"specification": str(specification_data)
|
|
}
|
|
response = requests.post(
|
|
url=events_url + "/v2.0/transforms",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
assert response.status_code == 200
|
|
print("POST /transforms success")
|
|
|
|
print("Test GET /transforms")
|
|
body = {}
|
|
|
|
response = requests.get(
|
|
url=events_url + "/v2.0/transforms",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
assert response.status_code == 200
|
|
print("GET /transforms success")
|
|
|
|
print("Test DELETE /transforms")
|
|
body = {}
|
|
|
|
response = requests.get(
|
|
url=events_url + "/v2.0/transforms",
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
transform_dict = json.loads(response.text)
|
|
transform_dict_id = transform_dict['elements'][0]['id']
|
|
response = requests.delete(
|
|
url=events_url + "/v2.0/transforms/{}".format(transform_dict_id),
|
|
data=json.dumps(body),
|
|
headers=headers)
|
|
assert response.status_code == 204
|
|
print("DELETE /transforms success")
|
|
|
|
test_stream_definition_post()
|
|
test_stream_definition_get()
|
|
test_stream_definition_delete()
|
|
test_events_get_all()
|
|
test_transforms()
|