Allow to view log

.github/workflows/code_style.yml

@@ -24,4 +24,4 @@ jobs:
       - name: "Run pylint"
         shell: bash
         run: |
-          pylint --disable R0401,W0511 --generated-member=requests.codes swattool
+          pylint --disable W0511 --generated-member=requests.codes swattool

swattool/logs.py

@@ -6,6 +6,8 @@
 import logging
 from typing import Optional
 
+import requests
+
 from .webrequests import Session
 
 logger = logging.getLogger(__name__)
@@ -15,12 +17,17 @@
 TYPHOON_API_URL = f"{TYPHOON_BASE_URL}/api/v2"
 
 
-def get_log_raw_url(buildid: int, stepid: int, logname: str) -> Optional[str]:
+def get_log_raw_url(buildid: int, stepnumber: int, logname: str
+                    ) -> Optional[str]:
     """Get URL of a raw log file, based on build and step ids."""
-    info_url = f"{TYPHOON_API_URL}/builds/{buildid}/steps/{stepid}" \
+    info_url = f"{TYPHOON_API_URL}/builds/{buildid}/steps/{stepnumber}" \
                f"/logs/{logname}"
 
-    info_data = Session().get(info_url)
+    try:
+        info_data = Session().get(info_url)
+    except requests.exceptions.HTTPError:
+        return None
+
     try:
         info_json_data = json.loads(info_data)
     except json.decoder.JSONDecodeError:

swattool/logsview.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+
+"""Swatbot review functions."""
+
+import logging
+import re
+import shutil
+from typing import Optional
+
+import requests
+from simple_term_menu import TerminalMenu  # type: ignore
+
+from . import logs
+from . import swatbuild
+from . import utils
+from .webrequests import Session
+
+logger = logging.getLogger(__name__)
+
+
+RESET = "\x1b[0m"
+RED = "\x1b[1;31m"
+GREEN = "\x1b[1;32m"
+YELLOW = "\x1b[1;33m"
+BLUE = "\x1b[1;34m"
+PURPLE = "\x1b[1;35m"
+CYAN = "\x1b[1;36m"
+WHITE = "\x1b[1;37m"
+
+
+def show_logs_menu(build: swatbuild.Build):
+    """Show a menu allowing to select log file to analyze."""
+    def get_failure_line(failure, logname):
+        return (failure.id, failure.stepnumber, failure.stepname, logname)
+    entries = [get_failure_line(failure, url)
+               for failure in build.failures.values()
+               for url in failure.urls]
+    default_line = get_failure_line(build.get_first_failure(), 'stdio')
+    entry = entries.index(default_line)
+    logs_menu = utils.tabulated_menu(entries, title="Log files",
+                                     cursor_index=entry)
+
+    while True:
+        newentry = logs_menu.show()
+        if newentry is None:
+            break
+
+        show_log_menu(build.id, entries[newentry][1], entries[newentry][3])
+
+
+def _format_log_line(linenum: int, text: str, colorized_line: Optional[int],
+                     highlight_lines: dict[int, tuple[str, str]]):
+    if linenum == colorized_line:
+        if linenum in highlight_lines:
+            linecolor = highlight_lines[linenum][1]
+        else:
+            linecolor = CYAN
+        text = f"{linecolor}{text}{RESET}"
+    elif linenum in highlight_lines:
+        pat = highlight_lines[linenum][0]
+        color = highlight_lines[linenum][1]
+        text = re.sub(pat, f"{color}{pat}{RESET}", text)
+    return text
+
+
+def _format_log_preview_line(linenum: int, text: str, colorized_line: int,
+                             highlight_lines: dict[int, tuple[str, str]]):
+    preview_text = text.replace('\t', '    ')
+    formatted_text = _format_log_line(linenum, preview_text, colorized_line,
+                                      highlight_lines)
+    return f"{linenum: 6d} {formatted_text}"
+
+
+def _get_preview_window(linenum: int, lines: list[str], preview_height: int
+                        ) -> tuple[int, int]:
+    start = max(0, linenum - int(preview_height / 4))
+    end = start + preview_height
+    if end >= len(lines):
+        end = len(lines)
+        start = max(0, end - preview_height)
+
+    return (start, end)
+
+
+def _format_log_preview(linenum: int, lines: list[str],
+                        highlight_lines: dict[int, tuple[str, str]],
+                        preview_height: int) -> str:
+    start, end = _get_preview_window(linenum, lines, preview_height)
+    lines = [_format_log_preview_line(i, t, linenum, highlight_lines)
+             for i, t in enumerate(lines[start: end], start=start + 1)]
+    return "\n".join(lines)
+
+
+def _get_log_highlights(loglines: list[str]) -> dict[int, tuple[str, str]]:
+    pats = [(re.compile(r"(.*\s|^)(?P<keyword>\S*error):", flags=re.I),
+             RED),
+            (re.compile(r"(.*\s|^)(?P<keyword>\S*warning):", flags=re.I),
+             YELLOW),
+            ]
+
+    highlight_lines = {}
+    for linenum, line in enumerate(loglines, start=1):
+        for (pat, color) in pats:
+            match = pat.match(line)
+            if match:
+                highlight_lines[linenum] = (match.group("keyword"), color)
+
+    return highlight_lines
+
+
+def _show_log(loglines: list[str], selected_line: Optional[int],
+              highlight_lines: dict[int, tuple[str, str]],
+              preview_height: Optional[int]):
+    colorlines = [_format_log_line(i, t, selected_line, highlight_lines)
+                  for i, t in enumerate(loglines, start=1)]
+
+    startline: Optional[int]
+    if selected_line and preview_height:
+        startline, _ = _get_preview_window(selected_line, loglines,
+                                           preview_height)
+        startline += 1  # Use line number, not line index
+    else:
+        startline = selected_line
+    utils.show_in_less("\n".join(colorlines), startline)
+
+
+def _load_log(buildid: int, stepnumber: int, logname: str
+              ) -> Optional[str]:
+    logurl = logs.get_log_raw_url(buildid, stepnumber, logname)
+    if not logurl:
+        logging.error("Failed to find log")
+        return None
+
+    try:
+        logdata = Session().get(logurl)
+    except requests.exceptions.ConnectionError:
+        logger.warning("Failed to download stdio log")
+        return None
+
+    return logdata
+
+
+def show_log_menu(buildid: int, stepnumber: int, logname: str):
+    """Analyze a failure log file."""
+    logdata = _load_log(buildid, stepnumber, logname)
+    if not logdata:
+        return
+
+    loglines = logdata.splitlines()
+    highlight_lines = _get_log_highlights(loglines)
+
+    entries = ["View entire log file|",
+               "View entire log file in default editor|",
+               *[f"On line {line: 6d}: {highlight_lines[line][0]}|{line}"
+                 for line in sorted(highlight_lines)]
+               ]
+
+    preview_size = 0.6
+    termheight = shutil.get_terminal_size((80, 20)).lines
+    preview_height = int(preview_size * termheight)
+
+    def preview(line):
+        return _format_log_preview(int(line), loglines, highlight_lines,
+                                   preview_height)
+
+    title = f"Log file: {logname} of build {buildid}, step {stepnumber}"
+    entry = 2
+    while True:
+        menu = TerminalMenu(entries, title=title, cursor_index=entry,
+                            preview_command=preview, preview_size=preview_size,
+                            raise_error_on_interrupt=True)
+        entry = menu.show()
+        if entry is None:
+            return
+
+        if entry == 0:
+            _show_log(loglines, None, highlight_lines, None)
+        elif entry == 1:
+            utils.launch_in_system_defaultshow_in_less(logdata)
+        else:
+            _, _, num = entries[entry].partition('|')
+            _show_log(loglines, int(num), highlight_lines, preview_height)

swattool/main.py

@@ -14,6 +14,7 @@
 from .bugzilla import Bugzilla
 from . import review
 from . import swatbot
+from . import swatbotrest
 from . import swatbuild
 from . import userdata
 from . import utils
@@ -36,7 +37,7 @@ def parse_filters(kwargs) -> dict[str, Any]:
     Parse filter values givean as program argument and generate a dictionary to
     be used with get_failure_infos().
     """
-    statuses = [swatbot.Status[s.upper()] for s in kwargs['status_filter']]
+    statuses = [swatbotrest.Status[s.upper()] for s in kwargs['status_filter']]
     tests = [re.compile(f"^{f}$") for f in kwargs['test_filter']]
     ignoretests = [re.compile(f"^{f}$") for f in kwargs['ignore_test_filter']]
     owners = [None if str(f).lower() == "none" else f
@@ -73,7 +74,7 @@ def main(verbose: int):
 @click.option('--password', '-p', prompt=True, hide_input=True)
 def login(user: str, password: str):
     """Login to the swatbot Django interface."""
-    swatbot.login(user, password)
+    swatbotrest.login(user, password)
 
 
 failures_list_options = [
@@ -97,7 +98,7 @@ def login(user: str, password: str):
     click.option('--ignore-test-filter', '-T', multiple=True,
                  help="Ignore some tests"),
     click.option('--status-filter', '-S', multiple=True,
-                 type=click.Choice([str(s) for s in swatbot.Status],
+                 type=click.Choice([str(s) for s in swatbotrest.Status],
                                    case_sensitive=False),
                  help="Only show some statuses"),
     click.option('--completed-after', '-A',
@@ -186,11 +187,11 @@ def show_pending_failures(refresh: str, open_url: str,
     logging.info("%s entries found (%s warnings, %s errors and %s cancelled)",
                  len(builds),
                  len([b for b in builds
-                      if b.status == swatbot.Status.WARNING]),
+                      if b.status == swatbotrest.Status.WARNING]),
                  len([b for b in builds
-                      if b.status == swatbot.Status.ERROR]),
+                      if b.status == swatbotrest.Status.ERROR]),
                  len([b for b in builds
-                      if b.status == swatbot.Status.CANCELLED]))
+                      if b.status == swatbotrest.Status.CANCELLED]))
 
 
 @main.command()
@@ -243,7 +244,7 @@ def publish_new_reviews(dry_run: bool):
         bugurl = None
 
         # Bug entry: need to also publish a new comment on bugzilla.
-        if status == swatbot.TriageStatus.BUG:
+        if status == swatbotrest.TriageStatus.BUG:
             bugid = int(comment)
             logs = [triage.extra['bugzilla-comment'] for triage in triages
                     if triage.failures]
@@ -261,7 +262,7 @@ def publish_new_reviews(dry_run: bool):
                             'to status %s (%s) with "%s"',
                             failureid, status, status.name.title(), comment)
                 if not dry_run:
-                    swatbot.publish_status(failureid, status, comment)
+                    swatbotrest.publish_status(failureid, status, comment)
 
     if not dry_run:
-        swatbot.invalidate_stepfailures_cache()
+        swatbotrest.invalidate_stepfailures_cache()