aboutsummaryrefslogtreecommitdiffstats
path: root/test/vpp_running.py
blob: f31a1327b4a0fec0c6a2a11a1fac12cf25fffb0b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#!/usr/bin/env python3

# Supporting module for running tests against a running VPP.
# This module is used by the test framework. Do not invoke this module
# directly for running tests against a running vpp. Use run.py for
# running all unit tests.

from glob import glob
import os
import sys
import subprocess
from config import config


def use_running(cls):
    """Update VPPTestCase to use running VPP's sock files & methods.

    Arguments:
    cls -- VPPTestCase Class
    """
    if config.running_vpp:
        if os.path.isdir(config.socket_dir):
            RunningVPP.socket_dir = config.socket_dir
        else:
            RunningVPP.socket_dir = RunningVPP.get_default_socket_dir()
        RunningVPP.get_set_vpp_sock_files()
        cls.get_stats_sock_path = RunningVPP.get_stats_sock_path
        cls.get_api_sock_path = RunningVPP.get_api_sock_path
        cls.run_vpp = RunningVPP.run_vpp
        cls.quit_vpp = RunningVPP.quit_vpp
        cls.vpp = RunningVPP
        cls.running_vpp = True
    return cls


class RunningVPP:
    api_sock = ""  # api_sock file path
    stats_sock = ""  # stats sock_file path
    socket_dir = ""  # running VPP's socket directory
    pid = None  # running VPP's pid
    returncode = None  # indicates to the framework that VPP is running

    @classmethod
    def get_stats_sock_path(cls):
        return cls.stats_sock

    @classmethod
    def get_api_sock_path(cls):
        return cls.api_sock

    @classmethod
    def run_vpp(cls):
        """VPP is already running -- skip this action."""
        pass

    @classmethod
    def quit_vpp(cls):
        """Indicate quitting to framework by setting returncode=1."""
        cls.returncode = 1

    @classmethod
    def terminate(cls):
        """Indicate termination to framework by setting returncode=1."""
        cls.returncode = 1

    @classmethod
    def get_default_socket_dir(cls):
        """Return running VPP's default socket directory.

        Default socket dir is:
           /var/run/user/${UID}/vpp  (or)
           /var/run/vpp, if VPP is started as a root user
        """
        if cls.is_running_vpp():
            vpp_user_id = (
                subprocess.check_output(["ps", "-o", "uid=", "-p", str(cls.pid)])
                .decode("utf-8")
                .strip()
            )
            if vpp_user_id == "0":
                return "/var/run/vpp"
            else:
                return f"/var/run/user/{vpp_user_id}/vpp"
        else:
            print(
                "Error: getting default socket dir, as "
                "a running VPP process could not be found"
            )
            sys.exit(1)

    @classmethod
    def get_set_vpp_sock_files(cls):
        """Look for *.sock files in the socket_dir and set cls attributes.

        Returns a tuple: (api_sock_file, stats_sock_file)
        Sets cls.api_sock and cls.stats_sock attributes
        """
        # Return if the sock files are already set
        if cls.api_sock and cls.stats_sock:
            return (cls.api_sock, cls.stats_sock)
        # Find running VPP's sock files in the socket dir
        if os.path.isdir(cls.socket_dir):
            if not cls.is_running_vpp():
                print(
                    "Error: The socket dir for a running VPP directory is, "
                    "set but a running VPP process could not be found"
                )
                sys.exit(1)
            sock_files = glob(os.path.join(cls.socket_dir + "/" + "*.sock"))
            for sock_file in sock_files:
                if "api.sock" in sock_file:
                    cls.api_sock = os.path.abspath(sock_file)
                elif "stats.sock" in sock_file:
                    cls.stats_sock = os.path.abspath(sock_file)
            if not cls.api_sock:
                print(
                    f"Error: Could not find a valid api.sock file "
                    f"in running VPP's socket directory {cls.socket_dir}"
                )
                sys.exit(1)
            if not cls.stats_sock:
                print(
                    f"Error: Could not find a valid stats.sock file "
                    f"in running VPP's socket directory {cls.socket_dir}"
                )
                sys.exit(1)
            return (cls.api_sock, cls.stats_sock)
        else:
            print("Error: The socket dir for a running VPP directory is unset")
            sys.exit(1)

    @classmethod
    def is_running_vpp(cls):
        """Return True if VPP's pid is visible else False."""
        vpp_pid = subprocess.Popen(
            ["pgrep", "-d,", "-x", "vpp_main"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
        )
        stdout, stderr = vpp_pid.communicate()
        cls.pid = int(stdout.split(",")[0]) if stdout else None
        return bool(cls.pid)

    @classmethod
    def poll(cls):
        """Return None to indicate that the process hasn't terminated."""
        return cls.returncode


if __name__ == "__main__":
    RunningVPP.socket_dir = RunningVPP.get_default_socket_dir()
    RunningVPP.get_set_vpp_sock_files()
    print(f"Running VPP's sock files")
    print(f"api_sock_file {RunningVPP.api_sock}")
    print(f"stats_sock_file {RunningVPP.stats_sock}")