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
|
import pytest, subprocess, re
from common import pppd_process
from packaging.version import Version
# pppd executable file name
@pytest.fixture()
def pppd(pytestconfig):
return pytestconfig.getoption("pppd")
# pppd configuration as string (should be redefined by specific test)
# all configs should contain "nodetach" option
@pytest.fixture()
def pppd_config():
return ""
# determines which plugin is required - pppoe.so (pppd 2.5.0+) or rp-pppoe.so (pppd <2.5.0)
def pppd_plugin_so(pppd):
command = [pppd, "--version"]
result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
pppd_version = Version(re.search(r'\d+\.\d+\.\d+', result.stdout + result.stderr).group())
ref_version = Version("2.5.0")
if pppd_version >= ref_version:
return "pppoe.so"
else:
return "rp-pppoe.so"
# pppd configuration as command line args
@pytest.fixture()
def pppd_args(pppd_config, pppd):
return ("plugin " + pppd_plugin_so(pppd) + "\n" + pppd_config).split()
# setup and teardown for tests that required running pppd (after accel-pppd)
@pytest.fixture()
def pppd_instance(accel_pppd_instance, veth_pair_netns, pppd, pppd_args):
# test setup:
print("pppd_instance: accel_pppd_instance = " + str(accel_pppd_instance))
is_started, pppd_thread, pppd_control = pppd_process.start(
veth_pair_netns["netns"],
pppd,
pppd_args,
)
# test execution:
yield {
"is_started": is_started,
"pppd_thread": pppd_thread,
"pppd_control": pppd_control,
}
# test teardown:
pppd_process.end(pppd_thread, pppd_control)
|
