Add support of deleting edge
Change-Id: I44493ed8a57ee339378b43ffcc2facb40629100c
diff --git a/edge-monitoring/VERSION b/edge-monitoring/VERSION
index 9e11b32..d15723f 100644
--- a/edge-monitoring/VERSION
+++ b/edge-monitoring/VERSION
@@ -1 +1 @@
-0.3.1
+0.3.2
diff --git a/edge-monitoring/edge_monitoring_server.py b/edge-monitoring/edge_monitoring_server.py
index 8ac3ea5..1bea10c 100755
--- a/edge-monitoring/edge_monitoring_server.py
+++ b/edge-monitoring/edge_monitoring_server.py
@@ -180,6 +180,20 @@
return jsonify({'edge': req_edge}), 201
+@app.route('/edges/<string:name>', methods=['DELETE'])
+def delete_edge(name):
+ print("delete edge request " + name)
+ result = False
+ for i in range(len(edges)):
+ if edges[i]['name'] == name:
+ del edges[i]
+ result = True
+ break
+ if not result:
+ abort(404)
+ return jsonify({'result': True})
+
+
if __name__ == '__main__':
if SECRET_ICAL_URL:
print(" * Starting maintenance calendar polling thread")
diff --git a/edge-monitoring/test_edge_monitoring_server.py b/edge-monitoring/test_edge_monitoring_server.py
index cf26f88..cb3a78a 100755
--- a/edge-monitoring/test_edge_monitoring_server.py
+++ b/edge-monitoring/test_edge_monitoring_server.py
@@ -18,6 +18,18 @@
import edge_monitoring_server as ems
import datetime
import pytz
+import json
+import time
+
+
+test_edge = {
+ 'name': 'production-edge-onf-menlo',
+ 'status': {
+ 'control_plane': 'connected',
+ 'user_plane': 'connected'
+ },
+ 'last_update': time.time()
+}
class MyEvent:
@@ -36,6 +48,9 @@
class TestEdgeMonitoringServer(unittest.TestCase):
+ def setUp(self):
+ self.app = ems.app.test_client()
+
def test_match_location(self):
event = MyEvent(location = "production-edge-onf-menlo, (Compute)-MP-1-Aether Production")
self.assertTrue(ems.is_my_event(event, "production-edge-onf-menlo"))
@@ -97,6 +112,29 @@
now = datetime.datetime(2020, 9, 3, 12, 0, tzinfo=pytz.utc)
self.assertFalse(ems.in_maintenance_window(events, "production-edge-onf-menlo", now))
+ def test_get_edges(self):
+ response = self.app.get('/edges')
+ data = json.loads(response.get_data(as_text=True))
+ self.assertEqual(len(data['edges']), 1)
+ self.assertEqual(data['edges'][0]['name'], 'production-edge-example')
+
+ def test_create_and_delete_edge(self):
+ response = self.app.post('/edges', json=test_edge)
+ data = json.loads(response.get_data(as_text=True))
+ self.assertEqual(data['edge']['name'], 'production-edge-onf-menlo')
+
+ response = self.app.get('/edges')
+ data = json.loads(response.get_data(as_text=True))
+ self.assertEqual(len(data['edges']), 2)
+
+ response = self.app.delete('/edges/production-edge-onf-menlo')
+ data = json.loads(response.get_data(as_text=True))
+ self.assertEqual(data['result'], True)
+ response = self.app.get('/edges', json=test_edge)
+
+ response = self.app.get('/edges')
+ data = json.loads(response.get_data(as_text=True))
+ self.assertEqual(len(data['edges']), 1)
if __name__ == '__main__':