client

#!/usr/bin/env python3

# based on code from git://github.com/openstack/nova.git
# nova/volume/nexenta/jsonrpc.py
#
# Copyright 2011 Nexenta Systems, Inc.
# All Rights Reserved.
#
#    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

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# Copyright 2012, Andy Grover <agrover@redhat.com>
#
# Test client to exercise targetd.
#

import json
import os
import time
import socket
import base64
import ssl

from test import testlib

from urllib.request import Request, urlopen

user = "admin"
password = "targetd"
host = "localhost"
port = 18700
path = "/targetrpc"
id_num = 1

# Allow us to test without SSL.
proto = os.getenv("TARGETD_UT_PROTO", "https")
print("CONNECT=%s" % proto)

if proto == "https":
    use_ssl = True
else:
    use_ssl = False

pools = ["vg-targetd/thin_pool", "zfs_targetd/block_pool"]
fs_pools = ["/mnt/btrfs", "/zfs_targetd/fs_pool"]


ctx = ssl.create_default_context(cafile=testlib.cert_file)


def jsonrequest(method, params=None):
    print("%s %s" % ("+" * 20, method))
    global id_num
    global use_ssl
    data = json.dumps(
        dict(id=id_num, method=method, params=params, jsonrpc="2.0")
    ).encode("utf-8")
    id_num += 1
    username_pass = "%s:%s" % (user, password)
    auth = base64.b64encode(username_pass.encode("utf-8")).decode("utf-8")
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Basic %s" % (auth,),
    }
    # print('Sending JSON data: %s' % data)
    if use_ssl:
        scheme = "https"
    else:
        scheme = "http"
    url = "%s://%s:%s%s" % (scheme, host, port, path)
    try:
        request = Request(url, data, headers)
        if use_ssl:
            response_obj = urlopen(request, context=ctx)
        else:
            response_obj = urlopen(request)
    except socket.error:
        print("error, retrying with SSL")
        url = "https://%s:%s%s" % (host, port, path)
        request = Request(url, data, headers)
        response_obj = urlopen(request, context=ctx)
        use_ssl = True
    response_data = response_obj.read().decode("utf-8")
    # print('Got response: %s' % response_data)
    response = json.loads(response_data)
    # Ensure we have version string
    assert response.get("jsonrpc") == "2.0"
    if response.get("error") is not None:
        if response["error"]["code"] <= 0:
            raise Exception(response["error"].get("message", ""))
        else:
            print("Invalid error code, should be negative!")
    else:
        return response.get("result")


results = jsonrequest("export_list")
for result in results:
    print(
        "export %s %s %s %s"
        % (
            str(result["initiator_wwn"]),
            str(result["pool"]),
            str(result["vol_name"]),
            str(result["lun"]),
        )
    )

# sys.exit(1)

results = jsonrequest("pool_list")
for result in results:
    print(
        "pool %s %s %s"
        % (str(result["name"]), str(result["size"]), str(result["free_size"]))
    )

# sys.exit(1)
failed = False

for pool in pools:
    print("Testing block pool: {0}".format(pool))

    results = jsonrequest("vol_list", dict(pool=pool))
    for result in results:
        print(
            "vol %s %s %s"
            % (str(result["name"]), str(result["size"]), str(result["uuid"]))
        )

    try:
        jsonrequest("vol_create", dict(pool=pool, name="test2", size=4 * 1024 * 1024))

        try:
            jsonrequest(
                "vol_copy", dict(pool=pool, vol_orig="test2", vol_new="test2-copy")
            )

            try:
                jsonrequest(
                    "export_create",
                    dict(
                        pool=pool,
                        vol="test2",
                        lun=5,
                        initiator_wwn="iqn.2006-03.com.wtf.ohyeah:666",
                    ),
                )

                print("waiting")
                time.sleep(5)
                results = jsonrequest("export_list")
                for result in results:
                    print(
                        "export %s %s %s %s %s"
                        % (
                            str(result["initiator_wwn"]),
                            str(result["pool"]),
                            str(result["vol_name"]),
                            str(result["lun"]),
                            str(result["vol_uuid"]),
                        )
                    )
                time.sleep(5)
                print("go!")

            finally:
                jsonrequest(
                    "export_destroy",
                    dict(
                        pool=pool,
                        vol="test2",
                        initiator_wwn="iqn.2006-03.com.wtf.ohyeah:666",
                    ),
                )

        finally:
            jsonrequest("vol_destroy", dict(pool=pool, name="test2-copy"))

        try:
            jsonrequest(
                "vol_copy",
                dict(
                    pool=pool,
                    vol_orig="test2",
                    vol_new="test2-copy-resize",
                    size=8 * 1024 * 1024,
                ),
            )

        finally:
            jsonrequest("vol_destroy", dict(pool=pool, name="test2-copy-resize"))

    finally:
        jsonrequest("vol_destroy", dict(pool=pool, name="test2"))
        print("done")

    print("block volumes left over:")
    results = jsonrequest("vol_list", dict(pool=pool))
    for fs in results:
        print("{pool}: {uuid} -> {name}".format(**fs))
        # No filesystems should be left
        failed = True

    if failed:
        print("Test failed")
        exit(1)

# ZFS fs tests
for fs_pool in fs_pools:
    fs_uuid = None

    print("Testing FS Pool {0}".format(fs_pool))
    try:
        jsonrequest("fs_create", dict(pool_name=fs_pool, name="test-fs", size_bytes=0))

        results = jsonrequest("fs_list")
        for fs in results:
            print("{pool}: {uuid} -> {name}".format(**fs))
            if fs["pool"] != fs_pool:
                failed = True
                raise Exception("unrecognized pool filesystem")
            fs_uuid = fs["uuid"]

            ss_uuid = None

            try:
                jsonrequest(
                    "fs_snapshot", dict(fs_uuid=fs_uuid, dest_ss_name="test-snapshot")
                )

                ss_results = jsonrequest("ss_list", dict(fs_uuid=fs_uuid))

                has_snapshot = False
                for ss in ss_results:
                    print("{uuid} -> {name} ({timestamp})".format(**ss))
                    if ss["name"] != "test-snapshot":
                        failed = True
                        raise Exception("unrecognized snapshot")

                    ss_uuid = ss["uuid"]
                    has_snapshot = True

                    clone_uuid = None

                    try:
                        jsonrequest(
                            "fs_clone",
                            dict(
                                fs_uuid=fs_uuid,
                                dest_fs_name="test-clone",
                                snapshot_id=ss_uuid,
                            ),
                        )

                        clone_results = jsonrequest("fs_list")

                        has_clone = False
                        for result in clone_results:
                            print("{pool}: {uuid} -> {name}".format(**result))
                            if result["name"] == "test-clone":
                                clone_uuid = result["uuid"]
                                has_clone = True
                        if not has_clone:
                            failed = True
                            raise Exception("test-clone not found")
                    finally:
                        if clone_uuid is not None:
                            jsonrequest("fs_destroy", dict(uuid=clone_uuid))

                if not has_snapshot:
                    failed = True
                    raise Exception("test-snapshot not found")

            finally:
                if ss_uuid is not None:
                    jsonrequest(
                        "fs_snapshot_delete", dict(fs_uuid=fs_uuid, ss_uuid=ss_uuid)
                    )
    finally:
        if fs_uuid is not None:
            jsonrequest("fs_destroy", dict(uuid=fs_uuid))

    print("fs'es left over:")
    results = jsonrequest("fs_list")
    for fs in results:
        print("{pool}: {uuid} -> {name}".format(**fs))
        # No filesystems should be left
        failed = True

    if failed:
        print("Test failed")
        exit(1)