lsquic

lsquic

Classes:

Name	Description
LsquicServiceManager

panther.plugins.services.iut.quic.lsquic.lsquic.LsquicServiceManager

LsquicServiceManager(service_config_to_test: LsquicConfig, service_type: str, protocol: ProtocolConfig, implementation_name: str)

Bases: IImplementationManager

Methods:

Name	Description
generate_compile_commands	This method generates and returns a list of compile commands.
generate_deployment_commands	Generates deployment commands and collects volume mappings based on service parameters.
generate_post_compile_commands	Generate a list of post-compile commands.
generate_post_run_commands	Generates post-run commands.
generate_pre_compile_commands	Generates a list of shell commands to be executed before compilation.
generate_pre_run_commands	Generates a list of pre-run commands.
generate_run_command	Generates the run command.
initialize_commands	Initializes and generates a dictionary of commands to be executed at different stages
prepare	Prepare the service manager for use.

Source code in panther/plugins/services/iut/quic/lsquic/lsquic.py

def __init__(
    self,
    service_config_to_test: LsquicConfig,
    service_type: str,
    protocol: ProtocolConfig,
    implementation_name: str,
):
    super().__init__(
        service_config_to_test, service_type, protocol, implementation_name
    )
    self.logger.debug(
        f"Initializing Lsquic service manager for '{implementation_name}'"
    )
    self.logger.debug(
        f"Loaded Lsquic configuration: {self.service_config_to_test}"
    )
    self.initialize_commands()

generate_compile_commands

generate_compile_commands() -> list[str]

This method generates and returns a list of compile commands. Generates compile commands.

Returns:

Name	Type	Description
list	list[str]	An empty list representing the compile commands.

Source code in panther/plugins/services/services_interface.py

def generate_compile_commands(self) -> list[str]:
    """
    This method generates and returns a list of compile commands.
    Generates compile commands.

    Returns:
        list: An empty list representing the compile commands.
    """
    return []

generate_deployment_commands

generate_deployment_commands() -> str

Generates deployment commands and collects volume mappings based on service parameters.

:param service_params: Parameters specific to the service. :param environment: The environment in which the services are being deployed. :return: A dictionary with service name as key and a dictionary containing command and volumes.

Source code in panther/plugins/services/iut/quic/lsquic/lsquic.py

def generate_deployment_commands(self) -> str:
    """
    Generates deployment commands and collects volume mappings based on service parameters.

    :param service_params: Parameters specific to the service.
    :param environment: The environment in which the services are being deployed.
    :return: A dictionary with service name as key and a dictionary containing command and volumes.
    """
    self.logger.debug(
        f"Generating deployment commands for service: {self.service_name} with service parameters: {self.service_config_to_test}"
    )
    # Create the command list

    self.logger.debug(f"Role: {self.role}, Version: {self.service_version}")

    # Determine if network interface parameters should be included based on environment
    include_interface = True

    # Build parameters for the command template
    # TODO ensure that the parameters are correctly set
    if self.role == RoleEnum.server:
        params = self.service_config_to_test.implementation.version.server
    # For the client, include target and message if available
    elif self.role == RoleEnum.client:
        params = self.service_config_to_test.implementation.version.client

    params["target"] = self.service_config_to_test.protocol.target

    self.logger.debug(f"Parameters for command template: {params}")
    self.logger.debug(f"Role: {self.role}")
    self.working_dir = params["binary"]["dir"]
    # Conditionally include network interface parameters
    if not include_interface:
        params["network"].pop("interface", None)
    else:
        # TODO add that in the Dockerfile
        subprocess.run(["bash", "generate_certificates.sh"])

    # Render the appropriate template
    try:
        template_name = f"{str(self.role.name)}_command.jinja"
        return super().render_commands(params, template_name)
    except Exception as e:
        self.logger.error(
            f"Failed to render command for service '{self.service_config_to_test.name}': {e}\n{traceback.format_exc()}"
        )
        raise e

generate_post_compile_commands

generate_post_compile_commands() -> list[str]

Generate a list of post-compile commands. This method returns an empty list of strings representing commands to be executed after the compilation process. Returns: List[str]: An empty list of post-compile commands.

Source code in panther/plugins/services/services_interface.py

def generate_post_compile_commands(self) -> list[str]:
    """
    Generate a list of post-compile commands.
    This method returns an empty list of strings representing commands
    to be executed after the compilation process.
    Returns:
        List[str]: An empty list of post-compile commands.
    """

    return []

generate_post_run_commands

generate_post_run_commands()

Generates post-run commands.

Source code in panther/plugins/services/iut/quic/lsquic/lsquic.py

def generate_post_run_commands(self):
    """
    Generates post-run commands.
    """
    return super().generate_post_run_commands() + [
        "cp /opt/lsquic/bin /app/logs/;"
    ]

generate_pre_compile_commands

generate_pre_compile_commands() -> list[str]

Generates a list of shell commands to be executed before compilation. Returns: list: A list of strings, each representing a shell command.

Source code in panther/plugins/services/services_interface.py

def generate_pre_compile_commands(self) -> list[str]:
    """
    Generates a list of shell commands to be executed before compilation.
    Returns:
        list: A list of strings, each representing a shell command.
    """

    return [
        "set -x;",
        "export PATH=$$PATH:$$ADDITIONAL_PATH;",
        "export PYTHONPATH=$$PYTHONPATH:$$ADDITIONAL_PYTHONPATH;",
        "env >> /app/logs/ivy_setup.log;",
    ]

generate_pre_run_commands

generate_pre_run_commands() -> list[str]

Generates a list of pre-run commands. This method returns an empty list of strings, which can be overridden by subclasses to provide specific pre-run commands required for their execution context. Returns: List[str]: An empty list of strings representing pre-run commands.

Source code in panther/plugins/services/services_interface.py

def generate_pre_run_commands(self) -> list[str]:
    """
    Generates a list of pre-run commands.
    This method returns an empty list of strings, which can be overridden by subclasses
    to provide specific pre-run commands required for their execution context.
    Returns:
        List[str]: An empty list of strings representing pre-run commands.
    """

    return []

generate_run_command

generate_run_command()

Generates the run command.

Source code in panther/plugins/services/iut/quic/lsquic/lsquic.py

def generate_run_command(self):
    """
    Generates the run command.
    """
    cmd_args = self.generate_deployment_commands()
    return {
        "working_dir": self.working_dir,
        "command_binary": (
            self.service_config_to_test.implementation.version.server.binary.name
            if self.role == RoleEnum.server
            else self.service_config_to_test.implementation.version.client.binary.name
        ),
        "command_args": cmd_args,
        "timeout": self.service_config_to_test.timeout,
        "command_env": {},
    }

initialize_commands

initialize_commands() -> dict

Initializes and generates a dictionary of commands to be executed at different stages of the process (pre-compile, compile, post-compile, pre-run, run, post-run).

The dictionary keys are

• "pre_compile_cmds": Commands to be executed before compilation.
• "compile_cmds": Commands to be executed during compilation.
• "post_compile_cmds": Commands to be executed after compilation.
• "pre_run_cmds": Commands to be executed before running.
• "run_cmd": Command to be executed to run the main process.
• "post_run_cmds": Commands to be executed after running.

Returns:

Name	Type	Description
dict	dict	A dictionary containing the commands for each stage.

Source code in panther/plugins/services/services_interface.py

@validate_cmd
def initialize_commands(self) -> dict:
    """
    Initializes and generates a dictionary of commands to be executed at different stages
    of the process (pre-compile, compile, post-compile, pre-run, run, post-run).

    The dictionary keys are:
        - "pre_compile_cmds": Commands to be executed before compilation.
        - "compile_cmds": Commands to be executed during compilation.
        - "post_compile_cmds": Commands to be executed after compilation.
        - "pre_run_cmds": Commands to be executed before running.
        - "run_cmd": Command to be executed to run the main process.
        - "post_run_cmds": Commands to be executed after running.

    Returns:
        dict: A dictionary containing the commands for each stage.
    """
    self.run_cmd = {
        "pre_compile_cmds": self.generate_pre_compile_commands(),
        "compile_cmds": self.generate_compile_commands(),
        "post_compile_cmds": self.generate_post_compile_commands(),
        "pre_run_cmds": self.generate_pre_run_commands(),
        "run_cmd": self.generate_run_command(),
        "post_run_cmds": self.generate_post_run_commands(),
    }
    self.logger.debug(f"Run commands: {self.run_cmd}")
    return self.run_cmd

prepare

prepare(plugin_loader: PluginLoader | None = None)

Prepare the service manager for use.

Source code in panther/plugins/services/iut/quic/lsquic/lsquic.py

def prepare(self, plugin_loader: PluginLoader | None = None):
    """
    Prepare the service manager for use.
    """
    self.logger.debug("Preparing Lsquic service manager...")
    plugin_loader.build_docker_image_from_path(
        Path(
            os.path.join(
                os.getcwd(),
                "panther",
                "plugins",
                "services",
                "Dockerfile",
            )
        ),
        "panther_base",
        "service",
    )
    plugin_loader.build_docker_image(
        self.get_implementation_name(),
        self.service_config_to_test.implementation.version,
    )