mesos_marathon_tests.py 文件源码

def test_app_communication(self):
        tapp = 'testapp2'
        # clean up any previous app by this name
        self.rt.delete_app(tapp)
        self.rt.create_hydra_app(name=tapp, app_path='hydra.selftest.agents.Test',
                                 app_args='5598 0',
                                 cpus=0.01, mem=32)
        taskip = self.rt.find_ip_uniqueapp(tapp)
        tasks = self.rt.get_app_tasks(tapp)
        self.assertTrue(len(tasks) == 1)
        self.assertTrue(len(tasks[0].ports) == 1)
        taskport = str(tasks[0].ports[0])

        pprint('task is launched at ip=' + taskip + ":" + taskport)
        # now send a message to this app to find out how it's doing
        zctx = zmq.Context()
        zsocket = zctx.socket(zmq.REQ)
        zsocket.connect("tcp://%s:%s" % (taskip, taskport))
        zsocket.send_string('ping')
        message = zsocket.recv().decode("utf-8")

        # stop and clean up
        self.rt.delete_app(tapp)
        self.assertEqual(message, 'pong')