fake_daqapp_rest

drunc.apps.fake_daqapp_rest

This is a fake DAQ application that doesn't do anything, but should talk in the same way to the run control.

It is primarily used for the testing of the Run Control.

Classes

AppCommand

Bases: Resource

Flask interface for the fake daq app.

Receives the commands from the command facility and passes them to the AppState to be executed, and sends the response back to the response listener.

Methods:

pass_daq_app(daq_app) classmethod

Interface to pass the daq_app instance to the Flask resource, since Flask doesn't allow to pass arguments to the resource constructor.

Source code in drunc/apps/fake_daqapp_rest.py

@classmethod
def pass_daq_app(cls, daq_app: AppState) -> type["AppCommand"]:
    """
    Interface to pass the daq_app instance to the Flask resource, since Flask
    doesn't allow to pass arguments to the resource constructor.
    """
    cls.daq_app = daq_app
    return cls

post()

Endpoint to receive commands from the command facility. The command data should be sent in a JSON format, with the following structure: { "id": "command_id", "entry_state": "state the app should be in to execute the command, or *", "exit_state": "state the app will be in after executing the command", "data": { optional parameters: "execution-time": "time the command should take to execute", "seg_fault": "app will exit with this code to simulate a failure", "throw": "app will throw an exception to simulate a failure" } }

Source code in drunc/apps/fake_daqapp_rest.py

def post(self) -> _CommandResult:
    """
    Endpoint to receive commands from the command facility. The command data should
    be sent in a JSON format, with the following structure:
    {
        "id": "command_id",
        "entry_state": "state the app should be in to execute the command, or *",
        "exit_state": "state the app will be in after executing the command",
        "data": { optional parameters:
            "execution-time": "time the command should take to execute",
            "seg_fault": "app will exit with this code to simulate a failure",
            "throw": "app will throw an exception to simulate a failure"
        }
    }
    """
    global app_state

    # Validate that the request contains JSON data
    try:
        data = request.get_json(force=True)
    except:
        return "Not a JSON command!\n", 406

    if not isinstance(data, dict):
        return "JSON command must be an object!\n", 406

    command_data = cast(CommandRequest, data)

    log = get_logger("fake_daqapp_rest.AppCommand")
    log.info(f"GET request with args: {data}")

    # Execute the command in a separate thread to not block the Flask app and to
    # allow concurrent command executions, since the app can receive multiple
    # commands while executing one command, and to allow the simulation of long
    # running commands without blocking the Flask app.
    thread = threading.Thread(
        target=self.daq_app.execute_command,
        kwargs={
            "req_data": cp.deepcopy(command_data),
            "answer_port": request.headers["X-Answer-Port"],
            "answer_host": request.headers.get("X-Answer-Host"),
            "remote_host": request.remote_addr,
        },
    )
    thread.start()

    return "Command received\n", 202

AppState(app_name)

Tracks state of the apps, and simulates the behaviour of daq_applicaitons with stateful commands.

Initialize the app state.

Parameters:

Name	Type	Description	Default
app_name	str	The name of the app, used for logging and in the responses	required

Returns:

Type	Description
	None

Source code in drunc/apps/fake_daqapp_rest.py

def __init__(self, app_name: str):
    """
    Initialize the app state.

    Args:
        app_name (str): The name of the app, used for logging and in the responses

    Returns:
        None

    Raises:
        None
    """
    self.appname = app_name
    self.state = "INITIAL"
    self.executing_command = False
    self.log = get_logger("fake_daqapp_rest.AppState", log_level="INFO")

Methods:

execute_command(req_data, answer_port, answer_host, remote_host)

Execute a command received from the command facility.

Parameters:

Name	Type	Description	Default
req_data	dict	The data received in the command, should contain at least the following keys: - id: The id of the command, used for logging and in the responses - entry_state: The state the app should be in to execute the command, or "*" to ignore the state - exit_state: The state the app will be in after executing the command - data: A dictionary with additional data for the command, can contain: - execution-time: An integer with the time the command should take to execute, in seconds - seg_fault: An integer that if present will cause the app to exit with that code - throw: If present, the app will throw an exception instead of executing the command	required
answer_port	str	The port to send the response to	required
answer_host	str | None	The host to send the response to, if None, the remote_host will be used	required
remote_host	str	The host that sent the command, used for logging and as a fallback for the answer_host	required

Returns:

Name	Type	Description
Response	Response	A Flask response object with the result of the command execution

Source code in drunc/apps/fake_daqapp_rest.py

def execute_command(
    self,
    req_data: CommandRequest,
    answer_port: str,
    answer_host: str | None,
    remote_host: str | None,
) -> Response:
    """
    Execute a command received from the command facility.

    Args:
        req_data (dict): The data received in the command, should contain at least
            the following keys:
            - id: The id of the command, used for logging and in the responses
            - entry_state: The state the app should be in to execute the command, or
                "*" to ignore the state
            - exit_state: The state the app will be in after executing the command
            - data: A dictionary with additional data for the command, can contain:
                - execution-time: An integer with the time the command should take
                    to execute, in seconds
                - seg_fault: An integer that if present will cause the app to exit
                    with that code
                - throw: If present, the app will throw an exception instead of
                    executing the command
        answer_port (str): The port to send the response to
        answer_host (str | None): The host to send the response to, if None, the
            remote_host will be used
        remote_host (str): The host that sent the command, used for logging and as a
            fallback for the answer_host

    Returns:
        Response: A Flask response object with the result of the command execution

    Raises:
        None
    """
    self.log.debug("Received command with the following data:")
    self.log.debug(f"{req_data=}")
    self.log.debug(f"{answer_port=}")
    self.log.debug(f"{answer_host=}")
    self.log.debug(f"{remote_host=}")

    # Construct the address to send the response to
    reply_address = (
        f"http://{answer_host}:{answer_port}/response"
        if answer_host
        else f"{remote_host or 'localhost'}:{answer_port}/response"
    )

    # Extract the relevant information from the command data
    entry_state = req_data["entry_state"]
    exit_state = req_data["exit_state"]
    command_id = req_data["id"]
    raw_data = req_data.get("data")
    data: _CmdData = raw_data if raw_data is not None else {}

    # If the app is already executing a command, it should not execute another one.
    # Send a response to the response listener indicating that it is busy
    if self.executing_command:
        response_txt = "Already executing a command!"
        self.log.info(
            "Application is already executing a command, cannot execute another "
            "one simultaneously."
        )
        self.send_response_to_response_listener(
            address=reply_address,
            txt=response_txt,
            success=False,
        )
        return Response("Already executing a command!\n", status=202)

    # Determine the time the command should take to execute. If not specified in the
    # data, it will be a random time between 1 and 5 seconds. We also determine a
    # random time for the worries, which is the time the app will wait before
    # failing the command in case of a seg_fault or throw, to simulate the time it
    # takes for the app to fail after starting the execution of the command.
    cmd_exec_time = data.get("execution-time", random.randint(1, 5))
    worries = random.randint(0, cmd_exec_time)

    # Validate that the app is in the correct state to execute the command. If not,
    # send a response to the response listener indicating that the command cannot
    # be executed due to the state of the app. The wildcard "*" can be used to
    # indicate that the command can be executed in any state.
    if entry_state != "*" and self.state != entry_state.upper():
        info = (
            f"DAQ Application is in state {self.state} and command {command_id} "
            f"requires to be in state {entry_state.upper()} to execute. Not "
            "executing."
        )
        self.log.info(info)
        self.send_response_to_response_listener(
            success=False,
            address=reply_address,
            txt=info,
        )
        return Response(f"{info}\n", status=202)

    # Execute the command, and mark the app as busy executing a command to prevent
    # concurrent executions.
    self.log.info(f"Executing {command_id}")
    self.executing_command = True

    # Failure testing through payload
    if data.get("seg_fault"):
        time.sleep(worries)
        app_execution_info = "<seeeeeeeeeg fauuuuuuuult message>"
        self.log.info(app_execution_info)
        self.send_response_to_response_listener(
            success=False,
            address=reply_address,
            txt=app_execution_info,
        )
        self.executing_command = False
        exit(data["seg_fault"])

    if data.get("throw"):
        time.sleep(worries)
        app_execution_info = (
            "This is an eRrOr, YoU hAvE bEeN vErY nAuGhTy (aka task failed "
            "successfully)"
        )
        self.log.info(app_execution_info)
        self.send_response_to_response_listener(
            success=False,
            address=reply_address,
            txt=app_execution_info,
        )
        self.executing_command = False
        return Response(f"{app_execution_info}\n", status=202)

    # FAILURE TESTING - CMD TIMEOUT
    # For testing purposes, we can delay the execution of the command to simulate a
    # long running command and test timeouts in the run control
    ft_fsm_timeout_raw = os.getenv("DRUNC_FT_FSM_CMD_TIMEOUT")
    ft_fsm_timeout = int(ft_fsm_timeout_raw) if ft_fsm_timeout_raw else None
    ft_fsm_timeout_cmd = os.getenv("DRUNC_FT_FSM_CMD_TIMEOUT_CMD")
    ft_fsm_timeout_app_name = os.getenv("DRUNC_FT_FSM_CMD_TIMEOUT_APP_NAME")
    if (
        ft_fsm_timeout
        and ft_fsm_timeout_cmd == command_id
        and ft_fsm_timeout_app_name == self.appname
    ):
        self.log.warning(
            f"Delaying execution of {command_id} in {ft_fsm_timeout_app_name} by "
            f"{ft_fsm_timeout} seconds"
        )
        time.sleep(ft_fsm_timeout)

    # FAILURE TESTING - CMD PROCESS DEATH
    # The following block simulates a failure of the app while executing a stateful
    # command. Thisserves uniquely to test the robustness of the Run Control when an
    # app exits upon running an applciation, and should not be used for any other
    # purpose.
    ft_fsm_death_cmd_raw = os.getenv("DRUNC_FT_FSM_CMD_DEATH_CMD", "")
    ft_fsm_death_cmd = False
    if ft_fsm_death_cmd_raw:
        ft_fsm_death_cmd = (
            ft_fsm_death_cmd_raw.strip('"').strip("'") == req_data["id"]
        )
    self.log.debug(f"{ft_fsm_death_cmd=}")

    ft_fsm_death_app_name_raw = os.getenv("DRUNC_FT_FSM_CMD_DEATH_APP_NAME", "")
    ft_fsm_death_app_name = False
    if ft_fsm_death_app_name_raw:
        ft_fsm_death_app_name = (
            ft_fsm_death_app_name_raw.strip('"').strip("'") == self.appname
        )
    self.log.debug(f"{ft_fsm_death_app_name=}")

    if ft_fsm_death_cmd and ft_fsm_death_app_name:
        self.log.debug("'Worries' sleeping prior to simulating process death")
        time.sleep(worries)
        self.log.warning(
            f"Simulating death of {self.appname} during FSM cmd execution"
        )
        # This requires a more agressive exit than sys.exit(), as this process is
        # running in a separate thread.
        os._exit(1)

    # "Execute" the command by sleeping for the determined time
    self.log.info(f"Sleeping for {cmd_exec_time} seconds")
    time.sleep(cmd_exec_time)

    # Notify command success
    app_execution_info = (
        f"Executed {command_id} successfully, after {cmd_exec_time} seconds"
    )
    self.log.info(app_execution_info)
    self.send_response_to_response_listener(
        success=True,
        address=reply_address,
        txt=app_execution_info,
    )

    # Update app state, and mark as not busy
    self.state = exit_state.upper()
    self.executing_command = False
    return Response(f"{app_execution_info}\n", status=202)

send_response_to_response_listener(address, txt, success=True, data=None)

Send a response to the response listener.

Parameters:

Name	Type	Description	Default
address	str	The address of the response listener	required
txt	str	The text to send in the response	required
success	bool	Whether the command was executed successfully or not	True
data	dict	Additional data to send in the response	None

Returns:

Type	Description
None	None

Source code in drunc/apps/fake_daqapp_rest.py

def send_response_to_response_listener(
    self,
    address: str,
    txt: str,
    success: bool = True,
    data: dict[str, object] | None = None,
) -> None:
    """
    Send a response to the response listener.

    Args:
        address (str): The address of the response listener
        txt (str): The text to send in the response
        success (bool): Whether the command was executed successfully or not
        data (dict): Additional data to send in the response

    Returns:
        None

    Raises:
        None
    """

    # The response is sent as a POST request to the response listener, with the
    # following contents in the body:
    data_to_send = {
        "success": success,
        "result": txt,
        "appname": self.appname,
        "data": data or {},
    }

    self.log.info(f"Sending RESPONSE to {address}, data: {data_to_send}")
    try:
        response = requests.post(
            address,
            json=data_to_send,
            headers={
                "Content-Type": "application/json",
            },
        )
        response.raise_for_status()
    except Exception as e:
        self.log.error("Couldn't send response to response listener")
        self.log.exception(e)

CommandRequest

Bases: TypedDict

Expected structure for incoming DAQ application command requests.

Functions:

get_address(hostname)

Gets a new address for the application, by finding an available port.

Parameters:

Name	Type	Description	Default
hostname	str	The hostname to use in the address	required

Returns:

Name	Type	Description
str	str	URI with the given hostname and a new available port

Source code in drunc/apps/fake_daqapp_rest.py

def get_address(hostname: str) -> str:
    """
    Gets a new address for the application, by finding an available port.

    Args:
        hostname: The hostname to use in the address

    Returns:
        str: URI with the given hostname and a new available port

    Raises:
        None
    """
    return f"rest://{hostname}:{get_new_port()}"

index()

Endpoint to check if the app is running, can be used in the tests to wait for the app to be ready before sending commands to it.

Returns:

Name	Type	Description
str	str	A string indicating the app is running.

Source code in drunc/apps/fake_daqapp_rest.py

def index() -> str:
    """
    Endpoint to check if the app is running, can be used in the tests to wait for the
    app to be ready before sending commands to it.

    Args:
        None

    Returns:
        str: A string indicating the app is running.

    Raises:
        None
    """
    return f"Fake DAQ app v{__version__}"

update_connectivity_service(name, connectivity_service, interval, url)

Function to continuously update the connectivity service with the address of the app, to simulate the behaviour of a real DAQ application that is continuously publishing its address to the connectivity service. This is necessary for the Run Control to be able to send commands to the app, since the Run Control gets the address of the app from the connectivity service. The function runs in a separate thread to not block the main thread of the app, which is running the Flask app to receive commands from the command facility.

Parameters:

Name	Type	Description	Default
name	str	The name of the publishing app	required
connectivity_service	ConnectivityServiceClient	the client to publish to the connectivity service	required
interval	int	Interval in seconds to update the connectivity service	required
url	str	The app address to publish to the connectivity service	required

Returns:

Type	Description
None	None

Source code in drunc/apps/fake_daqapp_rest.py

def update_connectivity_service(
    name: str, connectivity_service: ConnectivityServiceClient, interval: int, url: str
) -> None:
    """
    Function to continuously update the connectivity service with the address of the
    app, to simulate the behaviour of a real DAQ application that is continuously
    publishing its address to the connectivity service. This is necessary for the Run
    Control to be able to send commands to the app, since the Run Control gets the
    address of the app from the connectivity service. The function runs in a separate
    thread to not block the main thread of the app, which is running the Flask app to
    receive commands from the command facility.

    Args:
        name: The name of the publishing app
        connectivity_service: the client to publish to the connectivity service
        interval: Interval in seconds to update the connectivity service
        url: The app address to publish to the connectivity service

    Returns:
        None

    Raises:
        None
    """
    while True:
        connectivity_service.publish(
            name + "_control",
            url,
            "RunControlMessage",
        )
        time.sleep(interval)