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
| | #! /usr/bin/python3
# SPDX-License-Identifier: GPL-2.0-or-later
#
# Copyright Red Hat
# Author: David Gibson <david@gibson.dropbear.id.au>
"""
Test A Simple Socket Transport
pasta.py - Helpers for starting pasta
"""
import contextlib
import os.path
import tempfile
PASTA_BIN = './pasta'
class Pasta(contextlib.AbstractContextManager):
"""A managed pasta instance"""
def __init__(self, *, host, ns, opts):
self.host = host
self.ns = ns
self.opts = opts
self.proc = None
def __enter__(self):
self.tmpdir = tempfile.TemporaryDirectory()
piddir = self.tmpdir.__enter__()
pidfile = os.path.join(piddir, 'pasta.pid')
relpid = self.ns.relative_pid(self.host)
cmd = [f'{PASTA_BIN}', '-f', '-P', f'{pidfile}'] + list(self.opts) + \
[f'{relpid}']
self.proc = self.host.bg(*cmd)
self.proc.__enter__()
# Wait for the PID file to be written
pidstr = None
while not pidstr:
pidstr = self.host.output('cat', f'{pidfile}', check=False)
self.pid = int(pidstr)
return self
def __exit__(self, *exc_details):
try:
self.host.fg('kill', '-TERM', f'{self.pid}')
self.proc.__exit__(*exc_details)
finally:
self.tmpdir.__exit__(*exc_details)
|