summary |
shortlog |
log |
commit | commitdiff |
review |
tree
raw |
patch |
inline | side by side (from parent 1:
e491f4c)
Issue-ID: INF-308
Issue-ID: INF-334
Signed-off-by: Zhang Rong(Jon) <rong.zhang@windriver.com>
Change-Id: I5ce206ec9003ce1040e953f3b9495982ee3e8b49
API_HOST_EXTERNAL_FLOATING =
[API]
API_HOST_EXTERNAL_FLOATING =
[API]
+# support native_k8sapi,sol018,sol018_helmcli
+# if the value is black, then native_k8sapi will set by default
+DMS_SUPPORT_PROFILES =
return events_yaml_name
return "/configs/events.yaml"
return events_yaml_name
return "/configs/events.yaml"
-# get k8s host from env:
-
+# get k8s host from env:
def get_k8s_host():
k8s_host = os.environ.get("KUBERNETES_SERVICE_HOST")
if k8s_host is None:
raise Exception('Get k8s host failed.')
return k8s_host
def get_k8s_host():
k8s_host = os.environ.get("KUBERNETES_SERVICE_HOST")
if k8s_host is None:
raise Exception('Get k8s host failed.')
return k8s_host
-# get k8s host port from env:
-
+# get k8s host port from env:
def get_k8s_port():
k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", '443')
return k8s_port
def get_k8s_port():
k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", '443')
return k8s_port
def get_review_url():
try:
api = '/apis/authentication.k8s.io/v1/tokenreviews'
def get_review_url():
try:
api = '/apis/authentication.k8s.io/v1/tokenreviews'
except Exception:
raise Exception('Get k8s review url failed')
except Exception:
raise Exception('Get k8s review url failed')
def get_reviewer_token():
# token path default is below.
token_path = '/var/run/secrets/kubernetes.io/serviceaccount/token'
def get_reviewer_token():
# token path default is below.
token_path = '/var/run/secrets/kubernetes.io/serviceaccount/token'
def get_auth_provider():
return 'k8s'
def get_auth_provider():
return 'k8s'
+
+
+def get_dms_support_profiles():
+ profiles = config.conf.API.DMS_SUPPORT_PROFILES
+ profiles = profiles.replace("'", "").replace(
+ '"', "").replace('[', "").replace(']', "")
+ profiles = profiles.split(',')
+ if 'native_k8sapi' not in profiles:
+ profiles.append('native_k8sapi')
+ return profiles
import json
from o2common.domain.base import AgRoot, Serializer
import json
from o2common.domain.base import AgRoot, Serializer
-from o2common.config import conf as CONF
+from o2common.config import config, conf as CONF
# from dataclasses import dataclass
# from datetime import date
# from typing import Optional, List, Set
from .resource_type import ResourceKindEnum, ResourceTypeEnum
# from dataclasses import dataclass
# from datetime import date
# from typing import Optional, List, Set
from .resource_type import ResourceKindEnum, ResourceTypeEnum
-DeploymentManagerProfileDefault = 'default'
+DeploymentManagerProfileDefault = 'native_k8sapi'
DeploymentManagerProfileSOL018 = 'sol018'
DeploymentManagerProfileSOL018HelmCLI = 'sol018_helmcli'
DeploymentManagerProfileSOL018 = 'sol018'
DeploymentManagerProfileSOL018HelmCLI = 'sol018_helmcli'
d['profile'] = json.loads(d['profile'])
d['profileSupportList'] = [
DeploymentManagerProfileDefault,
d['profile'] = json.loads(d['profile'])
d['profileSupportList'] = [
DeploymentManagerProfileDefault,
- DeploymentManagerProfileSOL018,
- DeploymentManagerProfileSOL018HelmCLI,
+ profiles = config.get_dms_support_profiles()
+ for profile in profiles:
+ if profile == DeploymentManagerProfileSOL018:
+ d['profileSupportList'].append(profile)
+ elif profile == DeploymentManagerProfileSOL018HelmCLI:
+ d['profileSupportList'].append(profile)
)
first = res.first()
if first is None:
)
first = res.first()
if first is None:
- resourcetype_id = str(uuid.uuid4())
- uow.resource_types.add(ResourceType(
- resourcetype_id,
- 'pserver_mem', stxobj.type,
- resourcepool.oCloudId))
res_type_name = 'pserver_mem'
resourcetype_id = str(uuid.uuid3(
uuid.NAMESPACE_URL, res_type_name))
res_type_name = 'pserver_mem'
resourcetype_id = str(uuid.uuid3(
uuid.NAMESPACE_URL, res_type_name))
from o2common.service.messagebus import MessageBus
from o2common.views.pagination_route import link_header, PAGE_PARAM
from o2common.views.route_exception import NotFoundException
from o2common.service.messagebus import MessageBus
from o2common.views.pagination_route import link_header, PAGE_PARAM
from o2common.views.route_exception import NotFoundException
+from o2ims.domain import ocloud
from o2ims.views import ocloud_view
from o2ims.views.api_ns import api_ims_inventory as api_ims_inventory_v1
from o2ims.views.ocloud_dto import OcloudDTO, ResourceTypeDTO,\
from o2ims.views import ocloud_view
from o2ims.views.api_ns import api_ims_inventory as api_ims_inventory_v1
from o2ims.views.ocloud_dto import OcloudDTO, ResourceTypeDTO,\
@api_ims_inventory_v1.route("/v1/deploymentManagers/<deploymentManagerID>")
@api_ims_inventory_v1.param('deploymentManagerID',
'ID of the deployment manager')
@api_ims_inventory_v1.route("/v1/deploymentManagers/<deploymentManagerID>")
@api_ims_inventory_v1.param('deploymentManagerID',
'ID of the deployment manager')
-@api_ims_inventory_v1.param('profile', 'DMS profile: value supports "sol018"',
- _in='query')
+@api_ims_inventory_v1.param(
+ 'profile', 'DMS profile: value supports "native_k8sapi"',
+ _in='query')
@api_ims_inventory_v1.response(404, 'Deployment manager not found')
@api_ims_inventory_v1.param(
'all_fields',
@api_ims_inventory_v1.response(404, 'Deployment manager not found')
@api_ims_inventory_v1.param(
'all_fields',
args = parser.parse_args()
profile = (
args.profile if args.profile is not None and args.profile != ''
args = parser.parse_args()
profile = (
args.profile if args.profile is not None and args.profile != ''
+ else ocloud.DeploymentManagerProfileDefault)
result = ocloud_view.deployment_manager_one(
deploymentManagerID, bus.uow, profile)
result = ocloud_view.deployment_manager_one(
deploymentManagerID, bus.uow, profile)
+ if result is not None and result != "":
+ elif result == "":
+ raise NotFoundException(
+ "Profile {} doesn't support".format(
+ args.profile))
+
raise NotFoundException("Deployment manager {} doesn't exist".format(
deploymentManagerID))
raise NotFoundException("Deployment manager {} doesn't exist".format(
deploymentManagerID))
def deployment_manager_one(deploymentManagerId: str,
uow: unit_of_work.AbstractUnitOfWork,
def deployment_manager_one(deploymentManagerId: str,
uow: unit_of_work.AbstractUnitOfWork,
- profile: str = 'default'):
+ profile: str =
+ ocloud.DeploymentManagerProfileDefault):
profile = profile.lower()
with uow:
first = uow.deployment_managers.get(deploymentManagerId)
profile = profile.lower()
with uow:
first = uow.deployment_managers.get(deploymentManagerId)
profile_data = result.pop("profile", None)
result['profileName'] = profile
profile_data = result.pop("profile", None)
result['profileName'] = profile
+ profiles = config.get_dms_support_profiles()
+ if profile not in profiles:
+ return ""
- if ocloud.DeploymentManagerProfileDefault == profile:
- pass
- elif ocloud.DeploymentManagerProfileSOL018 == profile:
+ if ocloud.DeploymentManagerProfileDefault == profile \
+ or ocloud.DeploymentManagerProfileSOL018 == profile:
result['serviceUri'] = \
profile_data['cluster_api_endpoint']
result['profileData'] = profile_data
result['serviceUri'] = \
profile_data['cluster_api_endpoint']
result['profileData'] = profile_data
deploymentManagerId, profile_data)
result['profileData'] = helmcli_profile
else:
deploymentManagerId, profile_data)
result['profileData'] = helmcli_profile
else:
from o2ims.domain import ocloud, subscription_obj
from o2ims.domain import resource_type as rt
from o2ims.views import ocloud_view
from o2ims.domain import ocloud, subscription_obj
from o2ims.domain import resource_type as rt
from o2ims.views import ocloud_view
-from o2common.config import config
+from o2common.config import config, conf as CONF
+ CONF.API.DMS_SUPPORT_PROFILES = 'native_k8sapi,sol018,sol018_helmcli'
+ cluster_endpoint = "https://test_k8s:6443"
+ session.return_value.query.return_value.filter_by.return_value.first.\
+ return_value.serialize.return_value['profile'] = {
+ "cluster_api_endpoint": cluster_endpoint
+ }
+
# profile default
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow)
assert str(deployment_manager_res.get(
"deploymentManagerId")) == deployment_manager_id1
assert str(deployment_manager_res.get(
# profile default
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow)
assert str(deployment_manager_res.get(
"deploymentManagerId")) == deployment_manager_id1
assert str(deployment_manager_res.get(
- 'serviceUri')) == dms_endpoint
+ 'serviceUri')) == cluster_endpoint
assert deployment_manager_res.get('profile') is None
# profile sol018
profileName = ocloud.DeploymentManagerProfileSOL018
assert deployment_manager_res.get('profile') is None
# profile sol018
profileName = ocloud.DeploymentManagerProfileSOL018
- cluster_endpoint = "https://test_k8s:6443"
session.return_value.query.return_value.filter_by.return_value.first.\
return_value.serialize.return_value['profile'] = {
"cluster_api_endpoint": cluster_endpoint
session.return_value.query.return_value.filter_by.return_value.first.\
return_value.serialize.return_value['profile'] = {
"cluster_api_endpoint": cluster_endpoint
# profile wrong name
profileName = 'wrong_profile'
# profile wrong name
profileName = 'wrong_profile'
- session.return_value.query.return_value.filter_by.return_value.first.\
- return_value.serialize.return_value['profile'] = {
- "cluster_api_endpoint": cluster_endpoint
- }
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow, profile=profileName)
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow, profile=profileName)
- assert deployment_manager_res is None
+ assert deployment_manager_res == ""
def test_view_subscriptions(mock_uow):
def test_view_subscriptions(mock_uow):