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
| | #! /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
from typing import Iterator
from . import cmdsite, unshare
PASTA_BIN = '../pasta'
class _Pasta:
"""A running pasta instance"""
ns: unshare.Unshare
def __init__(self, ns: unshare.Unshare):
self.ns = ns
@contextlib.contextmanager
def pasta(host: cmdsite.CmdSite, ns: unshare.Unshare, *opts: str) \
-> Iterator[_Pasta]:
with tempfile.TemporaryDirectory() as piddir:
pidfile = os.path.join(piddir, 'pasta.pid')
relpid = ns.relative_pid(host)
cmd = [PASTA_BIN, '-f', '-P', pidfile] + list(opts) + [f'{relpid}']
with host.bg(*cmd):
# Wait for the PID file to be written
pidstr = None
while not pidstr:
pidstr = host.output('cat', pidfile, check=False)
pid = int(pidstr)
yield _Pasta(ns)
host.fg('kill', '-TERM', f'{pid}')
|