Skip to content

panther_runner ¤

Runner ¤

Runner(config, protocol_config, current_protocol, implems, executed_test=[])
Source code in panther/panther_worker/app/panther_runner/panther_runner.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self, config, protocol_config, current_protocol, implems, executed_test=[]
):
    # Setup logger
    self.log = logging.getLogger("panther-runner")
    self.log.setLevel(logging.INFO)
    # if (self.log.hasHandlers()):
    #     self.log.handlers.clear()
    self.log.addHandler(ch)
    self.log.propagate = False

    # Setup configuration
    self.log.info("START SETUP CONFIGURATION")
    self.current_protocol = current_protocol
    self.config = config
    self.log.info("SELECTED PROTOCOL: " + self.current_protocol)
    self.protocol_conf = protocol_config
    self.log.info("END SETUP PROTOCOL PARAMETERS")

    # TODO refactor
    self.iters = self.config["global_parameters"].getint(
        "iter"
    )  # Number of iteration per test           # TODO enforce in this file
    self.test_pattern = "*"  # Test to launch regex, * match all test # TODO

    self.extra_args = []  # TODO
    self.executed_tests = executed_test
    self.nb_test_to_execute = 0
    for mode in self.executed_tests.keys():
        self.nb_test_to_execute += len(self.executed_tests[mode])
    self.current_executed_test_count = 0

    self.implems = implems

    self.webapp_ip = socket.gethostbyname("panther-webapp")
    print(self.webapp_ip)
    print(self.nb_test_to_execute)
    print(self.nb_test_to_execute * self.config["global_parameters"].getint("iter"))
    # TODO make less general
    if (
        "quic_server_test_0rtt" in executed_test
        or "quic_client_test_0rtt" in executed_test
    ):
        self.bar_total_test = progressbar.ProgressBar(
            max_value=(self.nb_test_to_execute + 2)
            * self.config["global_parameters"].getint("iter")
        )
    else:
        self.bar_total_test = progressbar.ProgressBar(
            max_value=self.nb_test_to_execute
            * self.config["global_parameters"].getint("iter")
        )

copy_file ¤

copy_file(src, dst)

Copy a file from source to destination.

Parameters: src (str): Source file path. dst (str): Destination file path.

Source code in panther/panther_worker/app/panther_runner/panther_runner.py 
127
128
129
130
131
132
133
134
135
136
def copy_file(self, src, dst):
    """
    Copy a file from source to destination.

    Parameters:
    src (str): Source file path.
    dst (str): Destination file path.
    """
    self.log.info(f"Copy file: {src} to {dst}")
    shutil.copyfile(src, dst)

get_binary_details ¤

get_binary_details(implem, mode)

Get binary path and name for the given implementation and mode.

Parameters: implem (str): Implementation name. mode (str): Mode of the test (client/server).

Returns:

  • tuple –

    (binary_path, binary_name)

Source code in panther/panther_worker/app/panther_runner/panther_runner.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def get_binary_details(self, implem, mode):
    """
    Get binary path and name for the given implementation and mode.

    Parameters:
    implem (str): Implementation name.
    mode (str): Mode of the test (client/server).

    Returns:
        tuple: (binary_path, binary_name)
    """
    index = 0 if mode == "client" else 1
    binary_dir = self.implems[implem][index][implem]["binary-dir"]
    binary_name = self.implems[implem][index][implem]["binary-name"]

    binary_path = (
        binary_dir.replace("$IMPLEM_DIR", IMPLEM_DIR.replace("$PROT", self.current_protocol))
        .replace("$MODEL_DIR", MODEL_DIR)
        + "/"
        + binary_name.replace("$IMPLEM_DIR", IMPLEM_DIR.replace("$PROT", self.current_protocol))
        .replace("$MODEL_DIR", MODEL_DIR)
        .split(" ")[-1]
    )

    binary_name = binary_name.split("/")[-1].split(" ")[-1]
    return binary_path, binary_name

save_shadow_binaries ¤

save_shadow_binaries(implem, test, run_id)

Save shadow binaries for the given implementation and test.

Parameters: implem (str): Implementation name. test (object): Test object containing test details. run_id (int): Unique run identifier.

Source code in panther/panther_worker/app/panther_runner/panther_runner.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def save_shadow_binaries(self, implem, test, run_id):
    """
    Save shadow binaries for the given implementation and test.

    Parameters:
    implem (str): Implementation name.
    test (object): Test object containing test details.
    run_id (int): Unique run identifier.
    """
    if not self.config["net_parameters"].getboolean("shadow"):
        return

    self.log.info("Save shadow binaries:")

    try:
        binary_path, binary_name = self.get_binary_details(implem, test.mode)
        self.copy_file(binary_path, os.path.join(self.config["global_parameters"]["dir"], str(run_id), binary_name))

        test_path = os.path.join(self.config["global_parameters"]["build_dir"], test.name)
        dest_test_path = os.path.join(self.config["global_parameters"]["dir"], str(run_id), test.name)
        self.copy_file(test_path, dest_test_path)

    except Exception as e:
        self.log.error(f"Failed to save shadow binaries: {e}")