armour.py

import argparse
import json
import uuid

import pika
import requests
import time


def pretty_print_dict(d):
    return json.dumps(d, sort_keys=True, indent=4, separators=(',', ': '))


def http_is_ok(status):
    return 200 <= status <= 299


class SessionMockup(object):
    def __init__(self, users=None, session_name="my_session_name", base_url="http://localhost:5000"):
        self.users = users
        self.password = users
        self.session_name = session_name
        self.base_url = base_url

        # AMQP attributes
        self.amqp_url = "amqp://f-interop:Cowabunga@finterop-mq.noc.onelab.eu:443/%s" % self.session_name
        self.amqp_connection = None
        self.amqp_channel = None

    def connect_amqp(self):
        """

        :return:
        """
        self.amqp_connection = pika.BlockingConnection(pika.URLParameters(self.amqp_url))
        # Open the channel
        self.amqp_channel = self.amqp_connection.channel()
        # Enabled delivery confirmations
        self.amqp_channel.confirm_delivery()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def create_users(self):
        for username, password in zip(self.users, self.password):
            user_adding_request = requests.put("{base_url}/users/{username}".format(base_url=self.base_url,
                                                                                    username=username),
                                               data=json.dumps({"password": password}),
                                               headers={'Content-type': 'application/json',
                                                        'User-Agent': 'Requests'})
            # Return code is ok
            assert http_is_ok(user_adding_request.status_code)
            # User is available in the user list
            check_user_adding_request = requests.get("{base_url}/users".format(base_url=self.base_url))
            assert http_is_ok(check_user_adding_request.status_code)
            assert {'name': username} in json.loads(check_user_adding_request.text)

    def create_session(self):
        session_adding_request = requests.put("{base_url}/sessions/{session_name}".format(base_url=self.base_url,
                                                                                          session_name=self.session_name),
                                              data=json.dumps(self.settings),
                                              headers={'Content-type': 'application/json'})
        # Return code is ok
        assert http_is_ok(session_adding_request.status_code)
        # Session is available in the session list
        check_session_adding_request = requests.get("{base_url}/sessions/{session_name}".format(base_url=self.base_url,
                                                                                                session_name=self.session_name))
        session_data = json.loads(check_session_adding_request.text)
        assert http_is_ok(check_session_adding_request.status_code)
        assert session_data["status"] == "closed", pretty_print_dict(session_data)
        # TODO: assert session_name not in session_name["running"]

    def start_session(self):
        start_session_request = requests.patch(
            "{base_url}/sessions/{session_name}".format(session_name=self.session_name,
                                                        base_url=self.base_url),
            data=json.dumps({"status": "opened"}),
            headers={'Content-type': 'application/json'})
        # Return code is ok
        assert http_is_ok(start_session_request.status_code)
        # Session is in the right category
        check_session_started_request = requests.get(
            "{base_url}/sessions/{session_name}".format(session_name=self.session_name,
                                                        base_url=self.base_url))
        session_data = json.loads(check_session_started_request.text)
        assert http_is_ok(check_session_started_request.status_code)
        assert session_data["status"] == "opened", pretty_print_dict(session_data)

    def close_session(self):
        stop_session_request = requests.patch(
            "{base_url}/sessions/{session_name}".format(session_name=self.session_name,
                                                        base_url=self.base_url),
            data=json.dumps({"status": "closed"}),
            headers={'Content-type': 'application/json'})
        # Return code is ok
        assert http_is_ok(stop_session_request.status_code)
        # Session is in the right category
        check_session_stopped_request = requests.get(
            "{base_url}/sessions/{session_name}".format(session_name=self.session_name,
                                                        base_url=self.base_url))
        session_data = json.loads(check_session_stopped_request.text)
        assert http_is_ok(check_session_stopped_request.status_code)
        assert session_data["status"] == "closed", pretty_print_dict(session_data)

    def delete_session(self):
        deleting_session_request = requests.delete(
            "{base_url}/sessions/{session_name}".format(session_name=self.session_name,
                                                        base_url=self.base_url))
        # Return code is ok
        assert http_is_ok(deleting_session_request.status_code)
        # Session is not available in the session list
        check_deleting_session_request = requests.get("{base_url}/sessions".format(base_url=self.base_url))
        session_data = json.loads(check_deleting_session_request.text)
        assert http_is_ok(check_deleting_session_request.status_code)
        assert self.session_name not in session_data

    def delete_users(self):
        for username in self.users:
            deleting_user_request = requests.delete("{base_url}/users/%s" % username)
            # Return code is ok
            assert http_is_ok(deleting_user_request.status_code)
            # User is not available in the user list
            check_deleting_user_request = requests.get("{base_url}/users")
            assert http_is_ok(check_deleting_user_request.status_code)
            assert username not in json.loads(check_deleting_user_request.text)

    @property
    def settings(self):
        return {

            "testing_tools": "f-interop/6tisch",
            "tests": [],
            "users": list(self.users),
            "iuts": []
        }


def callback(channel, method, properties, body):
    print("We received: ")
    print(channel, method, properties, body)


configs = {
    "secjoin05": {"pan_id": "0xcafe",
                  "dag_root_mac_address": "00:12:4b:00:06:0d:84:64",
                  "join_proxy_mac_address": "00:12:4b:00:06:0d:9b:8c",
                  "pledge_mac_address": "00:12:4b:00:06:0d:9e:e1",
                  "wpan_channel": 26,
                  "test": "TD_6TiSCH_SECJOIN_05",
                  "assume_yes": True},
    "secjoin06": {"pan_id": "0xcafe",
                  "dag_root_mac_address": "00:12:4b:00:06:0d:84:64",
                  "join_proxy_mac_address": "00:12:4b:00:06:0d:9b:8c",
                  "pledge_mac_address": "00:12:4b:00:06:0d:9e:e1",
                  "wpan_channel": 26,
                  "test": "TD_6TiSCH_SECJOIN_06",
                  "assume_yes": True},
    "secjoin08": {"pan_id": "0xcafe",
                  "dag_root_mac_address": "00:12:4b:00:06:0d:84:64",
                  "join_proxy_mac_address": "00:12:4b:00:06:0d:9b:8c",
                  "pledge_mac_address": "00:12:4b:00:06:0d:9e:e1",
                  "wpan_channel": 26,
                  "test": "TD_6TiSCH_SECJOIN_08",
                  "assume_yes": True}
}

capture_path = {
    "secjoin05": "tests/test_6tisch/secjoin/TD_6TiSCH_SECJOIN_05/dumps/secjoin05_PASS_retransmission_of_join_request.pcapng",
    "secjoin06": "tests/test_6tisch/secjoin/TD_6TiSCH_SECJOIN_06/dumps/secjoin06_PASS.pcapng",
    "secjoin08": "tests/test_6tisch/secjoin/TD_6TiSCH_SECJOIN_08/dumps/secjoin08_PASS.pcapng"
}


def test_demo(test):
    """
    Simple demo script to instantiate the testing tool
    and start consuming messages
    :return:
    """
    user = str(uuid.uuid1())
    session = str(uuid.uuid1())
    queue = "armour_queue"
    exchange = "amq.topic"
    # You can bind on any routing key that you want
    # Routing key documentation: http://doc.f-interop.eu/
    routing_key = "control.testcase.verdict"
    print("User: %s" % user)
    print("Session: %s" % session)
    with SessionMockup(users=(user,),
                       session_name=session,
                       base_url="http://orchestrator.f-interop.eu") as s:
        s.create_users()
        print("User created on RMQ")
        s.create_session()
        print("Session created")
        s.connect_amqp()
        s.amqp_channel.queue_declare(queue=queue)
        print("Queue declared")
        s.amqp_channel.queue_bind(queue=queue,
                                  exchange=exchange,
                                  routing_key=routing_key)
        print("Queue bound")
        s.amqp_channel.basic_consume(consumer_callback=callback,
                                     queue=queue)
        print("Let's start the testing tool and wait a little for it to boot up")
        s.start_session()
        time.sleep(5)
        print("Let's send the configuration (test that we want, mac config...)")
        s.amqp_channel.basic_publish(
            exchange="amq.topic",
            routing_key=test,
            body=json.dumps(configs[test])
        )
        print("Configuration sent.")
        print("Let's send the PCAP")
        with open(capture_path[test], "rb") as f:
            s.amqp_channel.basic_publish(
                exchange="amq.topic",
                routing_key="data.pcapng",
                body=f.read()
            )
        print("Capture sent")

        print("Let's consume messages (We should see a verdict appears after a while)")
        s.amqp_channel.start_consuming()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('test',
                        help='foo help',
                        choices=['secjoin05', 'secjoin06', 'secjoin08'])
    args = parser.parse_args()
    test_demo(args.test)