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
|
#!/usr/bin/env python3
from podman import PodmanClient
from podman.domain.containers import Container
from podman import errors
from waiting import wait, TimeoutExpired
import pytest
import time
# The test requires podman root as seitan needs privileged capabilities
uri = "unix:///run/podman/podman.sock"
eater_image = "test-eater:latest"
seitan_image = "test-seitan:latest"
class ContainerConfig:
def __init__(self, seitan, eater,
exit_code_seitan, exit_code_eater, error_seitan, error_eater,
process, proc_args = []):
self.client = PodmanClient(base_url=uri)
self.exit_code_seitan = exit_code_seitan
self.exit_code_eater = exit_code_eater
self.error_seitan = error_seitan
self.error_eater = error_eater
# Create eater container
self.eater = self.client.containers.run(eater_image,
name = eater,
detach = True,
user = "1000",
tty = True,
command = ["/usr/bin/seitan-eater",
"-i",
"/var/run/test-filters/test.bpf",
"--",
process] + proc_args)
res = self.eater.inspect()
pid = (res["State"]["Pid"])
if not isinstance(pid, int):
sys.exit("pid isn't an integer:", pid)
# Create seitan container
self.seitan = self.client.containers.run(seitan_image,
name = seitan,
detach = True,
tty = True,
remove = True,
pid_mode = "host",
network_mode= "host",
privileged = True,
command = [ "/usr/bin/seitan",
"-p", str(pid),
# TODO: replace /dev/null with input file
"-i", "/dev/null"])
def wait_containers_creation(self):
self.eater.wait(interval="2s")
self.seitan.wait(interval="2s")
def wait_container_terminate(self):
def check_container_status(container: "Container"):
if(container.inspect()["State"]["Status"]) == "exited":
return True
return False
wait(lambda: check_container_status(self.eater), timeout_seconds=5)
def print_logs(self):
print("Output seitan:")
self.seitan.logs()
print("Output eater:")
self.eater.logs()
def check_results(self):
self.wait_containers_creation()
self.wait_container_terminate()
self.print_logs()
print("Got:", self.seitan.inspect() ["State"]["ExitCode"], "Expected:", self.exit_code_seitan)
assert (self.seitan.inspect() ["State"]["ExitCode"]) == self.exit_code_seitan
assert (self.eater.inspect()["State"]["ExitCode"]) == self.exit_code_eater
assert (self.seitan.inspect()["State"]["Error"]) == self.error_seitan
assert (self.eater.inspect()["State"]["Error"]) == self.error_eater
def restart_seitan(self):
self.seitan.restart()
def stop_eater(self):
self.eater.stop()
def clean_up_containers(containers = []):
with PodmanClient(base_url=uri) as client:
for c in containers:
client.containers.remove(c, force = True)
@pytest.fixture()
def seitan_container(request):
return "seitan_"+request.node.name
@pytest.fixture()
def eater_container(request):
return "eater_"+request.node.name
@pytest.fixture(autouse=True)
def setup(seitan_container, eater_container):
try:
clean_up_containers([seitan_container,eater_container])
except errors.exceptions.NotFound as e:
print("No previous container existing")
yield
try:
print("Delete ", seitan_container, eater_container)
clean_up_containers([seitan_container,eater_container])
except errors.exceptions.NotFound as e:
print("Containers already be removed")
def test_simple(seitan_container, eater_container):
test = ContainerConfig(seitan=seitan_container, eater=eater_container,
exit_code_seitan=0, exit_code_eater=0,
error_seitan="", error_eater= "",
process = "true")
test.check_results()
def test_restart_seitan(seitan_container, eater_container):
test = ContainerConfig(seitan=seitan_container, eater=eater_container,
exit_code_seitan=0, exit_code_eater=137,
error_seitan="", error_eater= "",
process = "sleep", proc_args = ["1000"])
# Give seitan some time to unblock the eater
# TODO: find a better way to detect that sleep has started
time.sleep(10)
test.restart_seitan()
test.stop_eater()
test.check_results()
|
