aspect/test/fn_cs_d2d_queue.py

# Tests client ability to query queues and read server queue messages.

import time

import flexitest

SERV_QUEUE_NAME = "server:homeserv_notif"

@flexitest.register
class Test(flexitest.Test):
    def __init__(self, ctx: flexitest.InitContext):
        ctx.set_env("basic2")

    def main(self, ctx: flexitest.RunContext):
        srpc = ctx.get_service("serv").create_rpc()
        c0rpc = ctx.get_service("cli0").create_rpc()
        c1rpc = ctx.get_service("cli1").create_rpc()

        c0did = c0rpc.aspc_status()["device_id"]
        c1did = c1rpc.aspc_status()["device_id"]
        print("c0 did", c0did)
        print("c1 did", c1did)

        mq01 = "d2d:%s,%s" % (c0did, c1did)
        mq10 = "d2d:%s,%s" % (c1did, c0did)
        print("queues", mq01, mq10)

        # Sending messages.
        print("sending first message")
        res = c0rpc.aspc_send_raw_d2d_msg(c1did, "01020304")
        print(res)
        print("sending second message")
        res = c1rpc.aspc_send_raw_d2d_msg(c0did, "05060708")
        print(res)

        res = c0rpc.aspc_list_device_queues()
        print("c0 queues", res)
        res = c1rpc.aspc_list_device_queues()
        print("c1 queues", res)