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
|
#!/usr/bin/python3
# SPDX-License-Identifier: MIT
"""mdbp backend wrapper via ssh"""
import argparse
import contextlib
import io
import json
import pathlib
import subprocess
import sys
import tarfile
import typing
from .common import JsonObject, buildjson, get_dsc_files, tar_add
def produce_request_tar(buildjsonobj: JsonObject,
fileobj: typing.IO[bytes]) -> None:
"""Write a tar file suitable for mdbp-streamapi into the given `fileobj`
based on the given `buildjsonobj`.
* An .output.directory is discarded.
* A referenced .dsc file and its components is included.
"""
sendjsonobj = buildjsonobj.copy()
sendjsonobj["output"] = sendjsonobj["output"].copy()
del sendjsonobj["output"]["directory"]
dscpath: typing.Optional[pathlib.Path]
try:
dscpath = pathlib.Path(buildjsonobj["input"]["source_package_path"])
except KeyError:
dscpath = None
else:
sendjsonobj["input"] = sendjsonobj["input"].copy()
sendjsonobj["input"]["source_package_path"] = dscpath.name
with tarfile.open(mode="w|", fileobj=fileobj) as tar:
info = tarfile.TarInfo("build.json")
sendjsonfile = io.BytesIO()
for chunk in json.JSONEncoder().iterencode(sendjsonobj):
sendjsonfile.write(chunk.encode("utf8"))
info.size = sendjsonfile.tell()
sendjsonfile.seek(0)
tar.addfile(info, sendjsonfile)
if dscpath:
for path in [dscpath] + get_dsc_files(dscpath):
tar_add(tar, path)
def main() -> None:
"""Entry point for mdbp-ssh backend"""
parser = argparse.ArgumentParser()
parser.add_argument("host", type=str)
parser.add_argument("command", nargs=argparse.REMAINDER)
args = parser.parse_args()
if len(args.command) < 2:
parser.error("missing command or json file")
build = buildjson(args.command.pop())
cmd = ["ssh", args.host, "mdbp-streamapi", *args.command]
with contextlib.ExitStack() as stack:
proc = stack.enter_context(
subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=sys.stdout if build["output"].get("log", True)
else subprocess.DEVNULL
),
)
assert proc.stdin is not None
produce_request_tar(build, proc.stdin)
proc.stdin.close()
exitcode = 0
try:
outtar = stack.enter_context(tarfile.open(fileobj=proc.stdout,
mode="r|"))
except tarfile.ReadError as err:
if str(err) != "empty file":
raise
exitcode = 1
else:
for member in outtar:
if "/" in member.name or not member.isfile():
raise ValueError("expected flat tar as output")
outtar.extract(member, build["output"]["directory"],
set_attrs=False)
sys.exit(proc.wait() or exitcode)
if __name__ == "__main__":
main()
|
