Method test_exec_interactive

src/python/cli_new/lib/cli/tests/task.py:142–187

Test for the task `exec()` sub-command, using `--interactive`.

(self)

    def test_exec_interactive(self):
        """
        Test for the task `exec()` sub-command, using `--interactive`.
        """
        # Launch a master, agent, and task.
        master = Master()
        master.launch()

        agent = Agent()
        agent.launch()

        task = Task({"command": "sleep 1000"})
        task.launch()

        try:
            wait_for_task(master, task.name, "TASK_RUNNING")
        except Exception as exception:
            raise CLIException(
                "Error waiting for task '{name}' to"
                " reach state '{state}': {error}"
                .format(name=task.name, state="TASK_RUNNING", error=exception))

        try:
            tasks = http.get_json(master.addr, "tasks")["tasks"]
        except Exception as exception:
            raise CLIException(
                "Could not get tasks from '/{endpoint}' on master: {error}"
                .format(endpoint="tasks", error=exception))

        self.assertEqual(type(tasks), list)
        self.assertEqual(len(tasks), 1)

        with open(LOREM_IPSUM) as text:
            returncode, stdout, stderr = exec_command(
                ["mesos", "task", "exec", "-i", tasks[0]["id"], "cat"],
                stdin=text)

        self.assertEqual(returncode, 0)
        with open(LOREM_IPSUM) as text:
            self.assertEqual(stdout, text.read())
        self.assertEqual(stderr, "")

        # Kill the task, agent, and master.
        task.kill()
        agent.kill()
        master.kill()


    def test_list(self):