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
|
#!/usr/bin/env python3
# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.
"""
Debusine integration tests.
Find the most recent task's debug logs and extract them to the specified path.
"""
import argparse
import sys
from pathlib import Path
from typing import Any, cast
import yaml
sys.path.append(str(Path(__file__).parent.parent))
from utils.client import Client # noqa: E402
def last_work_request() -> dict[str, Any]:
"""Return the most recent work request."""
result = Client.execute_command(
"work-request", "list", "--yaml", "--limit", "1"
)
try:
data = yaml.safe_load(result.stdout)
return cast(dict[str, Any], data[0])
except Exception as exc:
exc.add_note(f"Stdout: {result.stdout}")
exc.add_note(f"Stderr: {result.stderr}")
raise
def show_artifact(artifact_id: str | int) -> dict[str, Any]:
"""Return details for artifact_id."""
result = Client.execute_command("show-artifact", str(artifact_id))
return cast(dict[str, Any], yaml.safe_load(result.stdout))
def extract_last_debug_logs_to(destdir: Path) -> None:
"""Extract the debug logs for the last work request to destdir."""
work_request = last_work_request()
for artifact_id in work_request["artifacts"]:
artifact = show_artifact(artifact_id)
if artifact["category"] == "debusine:work-request-debug-logs":
Client.execute_command(
"download-artifact",
"--target-directory",
str(destdir),
artifact_id,
)
def main() -> None:
"""Parse arguments and run the job."""
p = argparse.ArgumentParser()
p.add_argument(
"destdir", type=Path, help="Directory to extract artifact into."
)
args = p.parse_args()
extract_last_debug_logs_to(args.destdir)
if __name__ == "__main__":
main()
|
