\r
## Running the tests\r
\r
+Prerequisite: in case of testing against real ocloud, download openrc file from ocloud dashboard, e.g. admin_openrc.sh\r
+\r
```sh\r
+source ./admin_openrc.sh\r
+export |grep OS_AUTH_URL\r
+export |grep OS_USERNAME\r
+export |grep OS_PASSWORD\r
docker-compose up -d\r
docker-compose run --rm --no-deps --entrypoint=pytest api /tests/unit /tests/integration\r
+\r
```\r
\r
## Tear down containers\r
- DB_PASSWORD=o2ims123
- REDIS_HOST=redis
- PYTHONDONTWRITEBYTECODE=1
+ - OS_AUTH_URL=${OS_AUTH_URL}
+ - OS_USERNAME=${OS_USERNAME}
+ - OS_PASSWORD=${OS_PASSWORD}
volumes:
- ./o2ims:/o2ims
- ./o2dms:/o2dms
- FLASK_APP=/o2ims/entrypoints/flask_application.py
- FLASK_DEBUG=1
- PYTHONUNBUFFERED=1
- - STX_AUTH_URL=http://192.168.204.1:5000/v3
- - STX_USERNAME=admin
- - STX_PASSWORD=password1
+ - OS_AUTH_URL=${OS_AUTH_URL}
+ - OS_USERNAME=${OS_USERNAME}
+ - OS_PASSWORD=${OS_PASSWORD}
volumes:
- ./o2ims:/o2ims
- ./o2dms:/o2dms
\r
# client talking to Stx standalone\r
\r
-from service.client.base_client import BaseClient\r
+from o2ims.service.client.base_client import BaseClient\r
from typing import List\r
# Optional, Set\r
from o2ims.domain import stx_object as ocloudModel\r
from o2ims import config\r
\r
+# from dcmanagerclient.api import client\r
+from cgtsclient.client import get_client\r
+import logging\r
+logger = logging.getLogger(__name__)\r
+\r
\r
class StxSaOcloudClient(BaseClient):\r
- def __init__(self):\r
+ def __init__(self, driver=None):\r
super().__init__()\r
- self.driver = StxSaClientImp()\r
+ self.driver = driver if driver else StxSaClientImp()\r
\r
# def list(self) -> List[ocloudModel.StxGenericModel]:\r
# return self._list()\r
# return self._get(id)\r
\r
def _get(self, id) -> ocloudModel.StxGenericModel:\r
- raise self.driver.getInstanceInfo()\r
+ return self.driver.getInstanceInfo()\r
\r
def _list(self):\r
return [self.driver.getInstanceInfo()]\r
def _list(self):\r
return self.driver.getK8sList()\r
\r
-# internal driver which implement client call to Stx Standalone instance\r
\r
-# from keystoneauth1.identity import v3\r
-# from keystoneauth1 import session\r
-# # from keystoneclient.v3 import ksclient\r
-# from starlingxclient.v3 import stxclient\r
+class StxPserverClient(BaseClient):\r
+ def __init__(self):\r
+ super().__init__()\r
+ self.driver = StxSaClientImp()\r
+\r
+ def _get(self, id) -> ocloudModel.StxGenericModel:\r
+ return self.driver.getPserver(id)\r
+\r
+ def _list(self) -> List[ocloudModel.StxGenericModel]:\r
+ return self.driver.getPserverList()\r
+\r
+# internal driver which implement client call to Stx Standalone instance\r
\r
\r
class StxSaClientImp(object):\r
- def __init__(self, access_info=None) -> None:\r
+ def __init__(self, stx_client=None):\r
super().__init__()\r
- self.access_info = access_info\r
- if self.access_info is None:\r
- self.access_info = config.get_stx_access_info()\r
- # self.auth = auth = v3.Password(\r
- # auth_url="http://example.com:5000/v3", username="admin",\r
- # password="password", project_name="admin",\r
- # user_domain_id="default", project_domain_id="default")\r
- # self.session = sess = session.Session(auth=auth)\r
- # # self.keystone = ksclient.Client(session=sess)\r
- # self.stx = stxclient.Client(session=sess)\r
+ self.stxclient = stx_client if stx_client else self.getStxClient()\r
+\r
+ def getStxClient():\r
+ os_client_args = config.get_stx_access_info()\r
+ config_client = get_client(**os_client_args)\r
+ return config_client\r
\r
def getInstanceInfo(self) -> ocloudModel.StxGenericModel:\r
- raise NotImplementedError\r
+ systems = self.stxclient.isystem.list()\r
+ logger.debug("systems:" + str(systems[0].to_dict()))\r
+ return ocloudModel.StxGenericModel(systems[0]) if systems else None\r
+\r
+ def getPserverList(self) -> List[ocloudModel.StxGenericModel]:\r
+ hosts = self.stxclient.ihost.list()\r
+ logger.debug("host 1:" + str(hosts[0].to_dict()))\r
+ return [ocloudModel.StxGenericModel(self._hostconverter(host))\r
+ for host in hosts if host]\r
+\r
+ def getPserver(self, id) -> ocloudModel.StxGenericModel:\r
+ host = self.stxclient.ihost.get(id)\r
+ logger.debug("host:" + str(host.to_dict()))\r
+ return ocloudModel.StxGenericModel(self._hostconverter(host))\r
\r
def getK8sList(self) -> List[ocloudModel.StxGenericModel]:\r
raise NotImplementedError\r
\r
def getK8sDetail(self, id) -> ocloudModel.StxGenericModel:\r
raise NotImplementedError\r
+\r
+ @staticmethod\r
+ def _hostconverter(host):\r
+ setattr(host, "name", host.hostname)\r
+ return host\r
\r
from sqlalchemy import (\r
Table,\r
- MetaData,\r
+ # MetaData,\r
Column,\r
# Integer,\r
String,\r
# from sqlalchemy.sql.expression import true\r
\r
from o2ims.domain import stx_object as ocloudModel\r
+from o2ims.adapter.orm import metadata\r
\r
logger = logging.getLogger(__name__)\r
\r
-metadata = MetaData()\r
+# metadata = MetaData()\r
\r
stxobject = Table(\r
"stxcache",\r
metadata,\r
Column("id", String(255), primary_key=True),\r
Column("name", String(255)),\r
- Column("lastupdate", DateTime),\r
+ Column("updatetime", DateTime),\r
+ Column("createtime", DateTime),\r
Column("content", String(255))\r
)\r
\r
# limitations under the License.
import os
+import sys
+import logging
def get_postgres_uri():
def get_stx_access_info():
- authurl = os.environ.get("STX_AUTH_URL", "http://192.168.204.1:5000/v3")
- username = os.environ.get("STX_USERNAME", "admin")
- pswd = os.environ.get("STX_PASSWORD", "passwd1")
- stx_access_info = (authurl, username, pswd)
- return stx_access_info
+ # authurl = os.environ.get("STX_AUTH_URL", "http://192.168.204.1:5000/v3")
+ # username = os.environ.get("STX_USERNAME", "admin")
+ # pswd = os.environ.get("STX_PASSWORD", "passwd1")
+ # stx_access_info = (authurl, username, pswd)
+ try:
+ client_args = dict(
+ auth_url=os.environ.get('OS_AUTH_URL',
+ "http://192.168.204.1:5000/v3"),
+ username=os.environ.get('OS_USERNAME', "admin"),
+ api_key=os.environ.get('OS_PASSWORD', "fakepasswd1"),
+ project_name=os.environ.get('OS_PROJECT_NAME', "admin"),
+ )
+ # dc_client_args = dict(
+ # auth_url=os.environ['OS_AUTH_URL'],
+ # username=os.environ['OS_USERNAME'],
+ # api_key=os.environ['OS_PASSWORD'],
+ # project_name=os.environ['OS_PROJECT_NAME'],
+ # user_domain_name=os.environ['OS_USER_DOMAIN_NAME'],
+ # project_domain_name=os.environ['OS_PROJECT_NAME'],
+ # project_domain_id=os.environ['OS_PROJECT_DOMAIN_ID']
+ # )
+ except KeyError:
+ logging.error('Please source your RC file before execution, '
+ 'e.g.: `source ~/downloads/admin-rc.sh`')
+ sys.exit(1)
+
+ os_client_args = {}
+ for key, val in client_args.items():
+ os_client_args['os_{key}'.format(key=key)] = val
+ os_client_args['os_password'] = os_client_args.pop('os_api_key')
+ os_client_args['os_region_name'] = 'RegionOne'
+ os_client_args['api_version'] = 1
+ return os_client_args
\r
# from dataclasses import dataclass\r
import datetime\r
+import json\r
\r
\r
class StxGenericModel:\r
- def __init__(self, id: str, name: str,\r
- lastupdate: datetime, content: str) -> None:\r
- self.id = id\r
- self.name = name\r
- self.lastupdate = lastupdate\r
- self.content = content\r
+ def __init__(self, api_response: dict) -> None:\r
+ self.id = api_response.uuid\r
+ self.content = json.dumps(api_response.to_dict())\r
+ self.updatetime = api_response.updated_at\r
+ self.createtime = api_response.created_at\r
+ self.name = api_response.name\r
+\r
+ # def __init__(self, id: str, name: str,\r
+ # lastupdate: datetime, content: str) -> None:\r
+ # self.id = id\r
+ # self.name = name\r
+ # self.lastupdate = lastupdate\r
+ # self.content = content\r
\r
def isChanged(self, updatetime: datetime) -> bool:\r
- return True if self.lastupdate > updatetime else False\r
+ return True if self.updatetime > updatetime else False\r
from tenacity import retry, stop_after_delay\r
\r
from o2ims.adapter.orm import metadata, start_o2ims_mappers\r
+from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers\r
from o2ims import config\r
\r
\r
@pytest.fixture\r
def mappers():\r
start_o2ims_mappers()\r
+ start_o2ims_stx_mappers()\r
yield\r
clear_mappers()\r
\r
+@pytest.fixture\r
+def fake_stx_client():\r
+ pass\r
\r
@retry(stop=stop_after_delay(10))\r
def wait_for_postgres_to_come_up(engine):\r
--- /dev/null
+# Copyright (C) 2021 Wind River Systems, Inc.\r
+#\r
+# Licensed under the Apache License, Version 2.0 (the "License");\r
+# you may not use this file except in compliance with the License.\r
+# You may obtain a copy of the License at\r
+#\r
+# http://www.apache.org/licenses/LICENSE-2.0\r
+#\r
+# Unless required by applicable law or agreed to in writing, software\r
+# distributed under the License is distributed on an "AS IS" BASIS,\r
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+# See the License for the specific language governing permissions and\r
+# limitations under the License.\r
+\r
+import sys\r
+import pytest\r
+from o2ims.adapter import ocloud_repository as repository\r
+from o2ims.domain import ocloud\r
+from o2ims import config\r
+import uuid\r
+from o2ims.adapter.clients.ocloud_sa_client import StxSaClientImp\r
+from cgtsclient.client import get_client\r
+\r
+import logging\r
+\r
+\r
+@pytest.fixture\r
+def real_stx_aio_client():\r
+ os_client_args = config.get_stx_access_info()\r
+ config_client = get_client(**os_client_args)\r
+ yield config_client\r
+\r
+# pytestmark = pytest.mark.usefixtures("mappers")\r
+\r
+\r
+def test_get_instanceinfo(real_stx_aio_client):\r
+ logger = logging.getLogger(__name__)\r
+ stxclientimp = StxSaClientImp(real_stx_aio_client)\r
+ assert stxclientimp is not None\r
+ systeminfo = stxclientimp.getInstanceInfo()\r
+ assert systeminfo is not None\r
+ assert systeminfo.id is not None\r
+ assert systeminfo.name is not None\r
+ assert systeminfo.content is not None\r
+\r
+\r
+def test_get_pserverlist(real_stx_aio_client):\r
+ stxSaClientImp = StxSaClientImp(real_stx_aio_client)\r
+ assert stxSaClientImp is not None\r
+ hosts = stxSaClientImp.getPserverList()\r
+ assert hosts is not None\r
+ assert len(hosts) > 0\r
+\r
+\r
+def test_get_pserver(real_stx_aio_client):\r
+ stxSaClientImp = StxSaClientImp(real_stx_aio_client)\r
+ assert stxSaClientImp is not None\r
+ hosts = stxSaClientImp.getPserverList()\r
+ assert hosts is not None\r
+ assert len(hosts) > 0\r
+ host1 = hosts[0]\r
+ host2 = stxSaClientImp.getPserver(host1.id)\r
+ assert host1 != host2\r
+ assert host1.id == host2.id\r
\r
[testenv:code]\r
commands =\r
- pytest tests\r
+ pytest tests/unit\r
\r
[testenv:nosetests]\r
commands =\r
- pytest tests\r
+ pytest tests/unit\r
Code Review / pti / o2.git / commitdiff