Method test_exec

src/python/cli_new/lib/cli/tests/task.py:49–94

Basic test for the task `exec()` sub-command.

(self)

    Test class for the task plugin.
    """
    def test_exec(self):
        """
        Basic test for the task `exec()` sub-command.
        """
        # Launch a master, agent, and task.
        master = Master()
        master.launch()

        agent = Agent()
        agent.launch()

        with open(LOREM_IPSUM) as text:
            content = text.read()
        command = "printf '{data}' > a.txt && sleep 1000".format(data=content)
        task = Task({"command": command})
        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)

        returncode, stdout, stderr = exec_command(
            ["mesos", "task", "exec", tasks[0]["id"], "cat", "a.txt"])

        self.assertEqual(returncode, 0)
        self.assertEqual(stdout, content)
        self.assertEqual(stderr, "")

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

    def test_exec_exit_status(self):
        """