test

Basic test Tasks for testing functionality.

Classes:

Name	Description
Test	Simplest test Task - runs a 10 iteration loop and returns a result.
TestSocket	Test Task which sends larger data to test socket IPC.
TestWriteOutput	Test Task which writes an output file.
TestReadOutput	Test Task which reads in a file. Can be used to test database access.

Test

Bases: Task

Simple test Task to ensure subprocess and pipe-based IPC work.

Source code in lute/tasks/test.py

class Test(Task):
    """Simple test Task to ensure subprocess and pipe-based IPC work."""

    def __init__(
        self, *, params: TestParameters, use_mpi: bool = False, row_ids=None
    ) -> None:
        super().__init__(params=params, use_mpi=use_mpi, row_ids=row_ids)

    def _run(self) -> None:
        self._task_parameters = cast(TestParameters, self._task_parameters)
        for i in range(10):
            time.sleep(1)
            msg: Message = Message(contents=f"Test message {i}")
            self._report_to_executor(msg)
        if self._task_parameters.throw_error:
            raise RuntimeError("Testing Error!")

    def _post_run(self) -> None:
        self._result.summary = "Test Finished."
        self._result.task_status = TaskStatus.COMPLETED
        time.sleep(0.1)

TestReadOutput

Bases: Task

Simple test Task to read in output from the test Task above.

Its pydantic model relies on a database access to retrieve the output file.

Source code in lute/tasks/test.py

class TestReadOutput(Task):
    """Simple test Task to read in output from the test Task above.

    Its pydantic model relies on a database access to retrieve the output file.
    """

    def __init__(
        self, *, params: TestReadOutputParameters, use_mpi: bool = False, row_ids=None
    ) -> None:
        super().__init__(params=params, use_mpi=use_mpi, row_ids=row_ids)

    def _run(self) -> None:
        self._task_parameters = cast(TestReadOutputParameters, self._task_parameters)
        _: np.ndarray = np.loadtxt(self._task_parameters.in_file, delimiter=",")
        self._report_to_executor(msg=Message(contents="Successfully loaded data!"))
        for i in range(5):
            time.sleep(1)

    def _post_run(self) -> None:
        super()._post_run()
        self._result.summary = "Was able to load data."
        self._result.payload = "This Task produces no output."
        self._result.task_status = TaskStatus.COMPLETED

TestRequest

Bases: Task

Simple test Task to try to send requests to and from via workflow manager.

Source code in lute/tasks/test.py

class TestRequest(Task):
    """Simple test Task to try to send requests to and from via workflow manager."""

    def __init__(
        self, *, params: TestReadOutputParameters, use_mpi: bool = False, row_ids=None
    ) -> None:
        super().__init__(params=params, use_mpi=use_mpi, row_ids=row_ids)

    def _parse_response(self, resp: Message) -> None:
        running_managed_tasks: List[str] = []
        running_tasks: List[str] = []
        if isinstance(resp.contents, dict):
            for managed_task_dict in resp.contents["managed_tasks"]:
                running_managed_tasks.append(managed_task_dict["name"])
                running_tasks.append(managed_task_dict["task"])

        n_tasks: int = len(running_managed_tasks)
        msg_str: str = f"TestRequest acknowledges that {n_tasks} Tasks are running\n"
        msg_str += "Their names are: " + ",".join(running_managed_tasks)

        self._report_to_executor(msg=Message(contents=msg_str))

    def _run(self) -> None:
        self._report_to_executor(Message(contents="Got to run"))
        time.sleep(1)

        # Format should be:
        # {
        #     "managed_tasks": [
        #         { "name": "...", "status": "...", "task": "...", OTHER KEYS MAYBE },
        #         { "name": "...", "status": "...", "task": "...", OTHER KEYS MAYBE },
        #     ]
        # }
        resp: Message = self.get_running_tasks()
        self._parse_response(resp=resp)
        time.sleep(1)

        resp = self.get_running_tasks()
        self._parse_response(resp=resp)
        time.sleep(1)

        resp = self.get_running_tasks()
        self._parse_response(resp=resp)
        time.sleep(1)

        self._report_to_executor(Message(contents="And done..."))

    def _post_run(self) -> None:
        super()._post_run()
        self._result.summary = "Was able to request via workflow manager.."
        self._result.payload = "This Task produces no output."
        self._result.task_status = TaskStatus.COMPLETED

TestSocket

Bases: Task

Simple test Task to ensure basic IPC over Unix sockets works.

Source code in lute/tasks/test.py

class TestSocket(Task):
    """Simple test Task to ensure basic IPC over Unix sockets works."""

    def __init__(
        self, *, params: TestSocketParameters, use_mpi: bool = False, row_ids=None
    ) -> None:
        super().__init__(params=params, use_mpi=use_mpi, row_ids=row_ids)

    def _run(self) -> None:
        self._task_parameters = cast(TestSocketParameters, self._task_parameters)
        for i in range(self._task_parameters.num_arrays):
            msg: Message = Message(contents=f"Sending array {i}")
            self._report_to_executor(msg)
            time.sleep(0.05)
            msg = Message(contents=np.random.rand(self._task_parameters.array_size))
            self._report_to_executor(msg)

    def _post_run(self) -> None:
        super()._post_run()
        self._task_parameters = cast(TestSocketParameters, self._task_parameters)
        self._result.summary = f"Sent {self._task_parameters.num_arrays} arrays"
        self._result.payload = np.random.rand(self._task_parameters.array_size)
        self._result.task_status = TaskStatus.COMPLETED

TestUnresponsive

Bases: Task

Simple test Task to check unresponsive Task cancellation.

Source code in lute/tasks/test.py

class TestUnresponsive(Task):
    """Simple test Task to check unresponsive Task cancellation."""

    def __init__(
        self, *, params: TestUnresponsiveParameters, use_mpi: bool = False, row_ids=None
    ) -> None:
        self._task_parameters: TestUnresponsiveParameters
        super().__init__(params=params, use_mpi=use_mpi, row_ids=row_ids)

    def _run(self) -> None:
        # Get the cancellation time (default is 900 s)
        # Hopefully someone set it so we don't wait 900 s...
        kill_time: float = float(os.getenv("LUTE_NO_HEARTBEAT_KILL", 900.0))
        wait_time: float = kill_time + 30.0
        msg: Message = Message(
            contents=(
                "TestUnresponsive has started! Going quiet! "
                f"You should kill me after {kill_time} s. I will wait {wait_time} s."
            )
        )
        self._report_to_executor(msg)
        time.sleep(wait_time)

        msg = Message(contents="If you get this, you didn't kill me! Why?!")
        self._report_to_executor(msg)

    def _post_run(self) -> None:
        self._result.summary = "TestUnresponsive Finished."
        self._result.task_status = TaskStatus.COMPLETED
        time.sleep(0.1)

TestWriteOutput

Bases: Task

Simple test Task to write output other Tasks depend on.

Source code in lute/tasks/test.py

class TestWriteOutput(Task):
    """Simple test Task to write output other Tasks depend on."""

    def __init__(
        self, *, params: TestWriteOutputParameters, use_mpi: bool = False, row_ids=None
    ) -> None:
        super().__init__(params=params, use_mpi=use_mpi, row_ids=row_ids)

    def _run(self) -> None:
        self._task_parameters = cast(TestWriteOutputParameters, self._task_parameters)
        for i in range(self._task_parameters.num_vals):
            # Doing some calculations...
            time.sleep(0.05)
            if i % 10 == 0:
                msg: Message = Message(contents=f"Processed {i+1} values!")
                self._report_to_executor(msg)

    def _post_run(self) -> None:
        super()._post_run()
        self._task_parameters = cast(TestWriteOutputParameters, self._task_parameters)
        work_dir: str = self._task_parameters.lute_config.work_dir
        out_file: str = f"{work_dir}/{self._task_parameters.outfile_name}"
        array: np.ndarray = np.random.rand(self._task_parameters.num_vals)
        np.savetxt(out_file, array, delimiter=",")
        self._result.summary = "Completed task successfully."
        self._result.payload = out_file
        self._result.task_status = TaskStatus.COMPLETED