conformance-tests/osm_client.py

# Copyright 2020 ETSI OSM
#
# 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 subprocess import run, PIPE
import requests
import time
import os

class OSMException(Exception):
    """A base class for MyProject exceptions."""

class ResourceException(OSMException):
    def __init__(self, *args, **kwargs):
        super().__init__(*args)
        self.message = kwargs.get('message')

class OSM:
    def __init__(self, osm_hostname, port):
        self.osm_hostname = osm_hostname
        self.port = port

    def run_command(self, command):
        p = run(["osm {}".format(command)], shell=True, stdout=PIPE, stderr=PIPE)
        if p.returncode != 0:
            print(p.stdout)
            print(p.stderr)
            raise ResourceException(message=p.stdout)
        else:
            return self.clean_output(p.stdout)

    def get_auth_token(self):

        r = requests.post("https://{}:{}/osm/admin/v1/tokens".format(self.osm_hostname, self.port), verify=False, headers={"Accept":"application/json"}, json={
                "username": "admin",
                "password": "admin",
                "project": "admin"
        })
        return r.json()["id"]

    def clean_output(self, output):
        return output.decode("utf-8").strip()

    def create_nsd(self, filename):
        return self.run_command("nsd-create {}".format(filename))

    def create_vnfd(self, filename):
        return self.run_command("vnfd-create {}".format(filename))

    def create_ns(self, name, nsd, vim):
        return self.run_command("ns-create --ns_name {} --nsd_name {} --vim_account {}".format(name, nsd, vim))

    def delete_nsd(self, id):
        return self.run_command("nsd-delete {}".format(id))

    def delete_vnfd(self, id):
        return self.run_command("vnfd-delete {}".format(id))

    def delete_ns(self, id):
        self.run_command("ns-delete {}".format(id))
        while (requests.get("https://{}:{}/osm/nslcm/v1/ns_instances/{}".format(self.osm_hostname, self.port, id), verify=False, headers={"Accept":"application/json", "Authorization":"Bearer {}".format(os.environ["AUTH_TOKEN"])}).status_code != 404):
            print("Deleting NS...")
            time.sleep(5)