from django.db import models
-class VnfInstModel(models.Model):
+class VnfRegModel(models.Model):
class Meta:
- db_table = 'GVNFM_VNFINST'
+ db_table = 'VNF_REG'
id = models.CharField(db_column='ID', primary_key=True, max_length=200)
-
+ ip = models.CharField(db_column='IP', max_length=200)
+ port = models.CharField(db_column='PORT', max_length=200)
+ username = models.CharField(db_column='USERNAME', max_length=255)
+ password = models.CharField(db_column='PASSWORD', max_length=255)
+
return args[old_key]
return def_val
+def set_opt_val(param, key, val):
+ if val or val is False:
+ param[key] = val
urlpatterns = [
url(r'^', include('mgr.samples.urls')),
+ url(r'^', include('mgr.vnfreg.urls')),
]
# regist to MSB when startup
--- /dev/null
+# Copyright 2017 ZTE Corporation.
+#
+# 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.
--- /dev/null
+# Copyright 2017 ZTE Corporation.
+#
+# 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 unittest
+import json
+from django.test import Client
+from rest_framework import status
+
+from mgr.pub.database.models import VnfRegModel
+
+class VnfRegTest(unittest.TestCase):
+ def setUp(self):
+ self.client = Client()
+ VnfRegModel.objects.filter().delete()
+ self.vnfInst1 = {
+ "vnfInstId": "1",
+ "ip": "192.168.0.1",
+ "port": "2324",
+ "username": "admin",
+ "password": "admin123"
+ }
+ self.vnfInst1_new = {
+ "vnfInstId": "1",
+ "ip": "192.168.0.2",
+ "port": "2325",
+ "username": "admin1",
+ "password": "admin1234"
+ }
+
+ def tearDown(self):
+ pass
+
+ def test_add_vnf_normal(self):
+ response = self.client.post("/openoapi/vnfmgr/v1/vnfs", self.vnfInst1, format='json')
+ self.assertEqual(status.HTTP_201_CREATED, response.status_code, response.content)
+ vnfs = VnfRegModel.objects.filter()
+ self.assertEqual(1, len(vnfs))
+ vnfInstActual = {
+ "vnfInstId": vnfs[0].id,
+ "ip": vnfs[0].ip,
+ "port": vnfs[0].port,
+ "username": vnfs[0].username,
+ "password": vnfs[0].password
+ }
+ self.assertEqual(self.vnfInst1, vnfInstActual)
+
+ def test_add_vnf_when_duplicate(self):
+ self.client.post("/openoapi/vnfmgr/v1/vnfs", self.vnfInst1, format='json')
+ response = self.client.post("/openoapi/vnfmgr/v1/vnfs", self.vnfInst1, format='json')
+ self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code, response.content)
+ self.assertEqual({'error': "Vnf(1) already exists."}, json.loads(response.content))
+
+ def test_set_vnf_normal(self):
+ self.client.post("/openoapi/vnfmgr/v1/vnfs", self.vnfInst1, format='json')
+ response = self.client.put("/openoapi/vnfmgr/v1/vnfs/1",
+ json.dumps(self.vnfInst1_new), content_type='application/json')
+ self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
+ vnfs = VnfRegModel.objects.filter()
+ self.assertEqual(1, len(vnfs))
+ vnfInstActual = {
+ "vnfInstId": vnfs[0].id,
+ "ip": vnfs[0].ip,
+ "port": vnfs[0].port,
+ "username": vnfs[0].username,
+ "password": vnfs[0].password
+ }
+ self.assertEqual(self.vnfInst1_new, vnfInstActual)
+
+ def test_set_vnf_when_not_exist(self):
+ response = self.client.put("/openoapi/vnfmgr/v1/vnfs/1",
+ json.dumps(self.vnfInst1_new), content_type='application/json')
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code, response.content)
+ self.assertEqual({'error': "Vnf(1) does not exist."}, json.loads(response.content))
+
+ def test_get_vnf_normal(self):
+ self.client.post("/openoapi/vnfmgr/v1/vnfs", self.vnfInst1, format='json')
+ response = self.client.get("/openoapi/vnfmgr/v1/vnfs/1")
+ self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
+ self.assertEqual(self.vnfInst1, json.loads(response.content))
+
+ def test_get_vnf_when_not_exist(self):
+ response = self.client.get("/openoapi/vnfmgr/v1/vnfs/1")
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code, response.content)
+ self.assertEqual({'error': "Vnf(1) does not exist."}, json.loads(response.content))
+
+ def test_del_vnf_normal(self):
+ self.client.post("/openoapi/vnfmgr/v1/vnfs", self.vnfInst1, format='json')
+ response = self.client.delete("/openoapi/vnfmgr/v1/vnfs/1")
+ self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code, response.content)
+
+ def test_del_vnf_when_not_exist(self):
+ response = self.client.delete("/openoapi/vnfmgr/v1/vnfs/1")
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code, response.content)
+ self.assertEqual({'error': "Vnf(1) does not exist."}, json.loads(response.content))
+
+ def test_url(self):
+ pass
+ #resp_data = json.loads(response.content)
+ #self.assertEqual({"status": "active"}, resp_data)
--- /dev/null
+# Copyright 2017 ZTE Corporation.
+#
+# 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.
+
+from django.conf.urls import patterns, url
+from rest_framework.urlpatterns import format_suffix_patterns
+from mgr.vnfreg import views
+
+urlpatterns = [
+ url(r'^openoapi/vnfmgr/v1/vnfs$', views.add_vnf, name='add_vnf'),
+ url(r'^openoapi/vnfmgr/v1/vnfs/(?P<vnfInstId>[0-9a-zA-Z\-\_]+)$', views.access_vnf, name='access_vnf'),
+ url(r'^openoapi/vnfmgr/v1/configuration$', views.vnf_config, name='vnf_config'),
+]
+
+urlpatterns = format_suffix_patterns(urlpatterns)
+
--- /dev/null
+# Copyright 2017 ZTE Corporation.
+#
+# 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 logging
+
+from rest_framework import status
+from rest_framework.decorators import api_view
+from rest_framework.response import Response
+
+from mgr.pub.utils.values import ignore_case_get
+from mgr.pub.utils.syscomm import fun_name
+from mgr.pub.database.models import VnfRegModel
+
+logger = logging.getLogger(__name__)
+
+@api_view(http_method_names=['POST'])
+def add_vnf(request, *args, **kwargs):
+ logger.info("Enter %s, data is %s", fun_name(), request.data)
+ vnf_inst_id = ignore_case_get(request.data, "vnfInstId")
+ try:
+ if VnfRegModel.objects.filter(id=vnf_inst_id):
+ raise Exception("Vnf(%s) already exists." % vnf_inst_id)
+ VnfRegModel(
+ id=vnf_inst_id,
+ ip=ignore_case_get(request.data, "ip"),
+ port=ignore_case_get(request.data, "port"),
+ username=ignore_case_get(request.data, "username"),
+ password=ignore_case_get(request.data, "password")).save()
+ except Exception as e:
+ return Response(data={'error': e.message}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+ return Response(data={"vnfInstId": vnf_inst_id}, status=status.HTTP_201_CREATED)
+
+
+@api_view(http_method_names=['GET', 'PUT', 'DELETE'])
+def access_vnf(request, *args, **kwargs):
+ vnf_inst_id = ignore_case_get(kwargs, "vnfInstId")
+ logger.info("Enter %s, method is %s, ", fun_name(), request.method)
+ logger.info("vnfInstId is %s, data is %s", vnf_inst_id, request.data)
+ ret, normal_status = None, None
+ try:
+ vnf = VnfRegModel.objects.filter(id=vnf_inst_id)
+ if not vnf:
+ err_msg = "Vnf(%s) does not exist." % vnf_inst_id
+ return Response(data={'error': err_msg}, status=status.HTTP_404_NOT_FOUND)
+ if request.method == 'GET':
+ ret = {
+ "vnfInstId": vnf_inst_id,
+ "ip": vnf[0].ip,
+ "port": vnf[0].port,
+ "username": vnf[0].username,
+ "password": vnf[0].password
+ }
+ normal_status = status.HTTP_200_OK
+ elif request.method == 'PUT':
+ ip = ignore_case_get(request.data, "ip")
+ port = ignore_case_get(request.data, "port")
+ username = ignore_case_get(request.data, "username")
+ password = ignore_case_get(request.data, "password")
+ if ip:
+ vnf[0].ip = ip
+ if port:
+ vnf[0].port = port
+ if username:
+ vnf[0].username = username
+ if password:
+ vnf[0].password = password
+ vnf[0].save()
+ ret = {}
+ normal_status = status.HTTP_202_ACCEPTED
+ else:
+ vnf.delete()
+ ret = {}
+ normal_status = status.HTTP_204_NO_CONTENT
+ except Exception as e:
+ return Response(data={'error': e.message}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+ return Response(data=ret, status=normal_status)
+
+@api_view(http_method_names=['POST'])
+def vnf_config(request, *args, **kwargs):
+ logger.info("Enter %s, data is %s", fun_name(), request.data)
+
\ No newline at end of file
Code Review / vfc / gvnfm / vnfmgr.git / commitdiff