271 lines
9.6 KiB
Python
271 lines
9.6 KiB
Python
# Copyright 2023 (c) Anna Schumaker.
|
|
"""Our Testcase Gio.ListModel."""
|
|
import sqlite3
|
|
import typing
|
|
from gi.repository import GObject
|
|
from gi.repository import Gio
|
|
from gi.repository import Gtk
|
|
from .. import sqlite
|
|
|
|
|
|
class XunitCell(GObject.GObject):
|
|
"""Holds a single value for a single Xunit."""
|
|
|
|
name = GObject.Property(type=str)
|
|
|
|
def __str__(self) -> str:
|
|
"""Get a string representation of this XunitCell."""
|
|
return self.name
|
|
|
|
|
|
class XunitRow(GObject.GObject):
|
|
"""Collects results for a single row across multiple Xunits."""
|
|
|
|
name = GObject.Property(type=str)
|
|
|
|
def __init__(self, name: str) -> None:
|
|
"""Initialize an XunitRow."""
|
|
super().__init__(name=name)
|
|
self.__xunits = {}
|
|
|
|
def __getitem__(self, xunit: str) -> XunitCell | None:
|
|
"""Get the value of a specific Xunit."""
|
|
return self.__xunits.get(xunit)
|
|
|
|
def __lt__(self, rhs: typing.Self) -> bool:
|
|
"""Compare the names of two XunitRows."""
|
|
return self.name < rhs.name
|
|
|
|
def add_xunit(self, name: str, *args, **kwargs) -> None:
|
|
"""Add an XunitCell to the XunitRow."""
|
|
self.__xunits[name] = self.do_make_xunit(name, *args, **kwargs)
|
|
|
|
def do_make_xunit(self, name: str) -> XunitCell:
|
|
"""Create and return a new XunitCell."""
|
|
return XunitCell(name=name)
|
|
|
|
def get_results(self) -> set[str]:
|
|
"""Get a set of results for each added xunit."""
|
|
return {str(xunit) for xunit in self.__xunits.values()}
|
|
|
|
|
|
class XunitList(GObject.GObject, Gio.ListModel):
|
|
"""A list of XunitRows for a specific Xfstests Run."""
|
|
|
|
runid = GObject.Property(type=int)
|
|
n_items = GObject.Property(type=int)
|
|
|
|
def __init__(self, sql: sqlite.Connection, runid: int) -> None:
|
|
"""Initialize an XunitList."""
|
|
super().__init__(runid=runid)
|
|
self.__xunits = set()
|
|
|
|
rows = {}
|
|
for row in self.do_query(sql).fetchall():
|
|
self.do_parse(rows, row)
|
|
self.__xunits.add(row["xunit"])
|
|
|
|
self.__items = sorted(rows.values())
|
|
self.n_items = len(self.__items)
|
|
|
|
def do_get_item_type(self) -> GObject.GType:
|
|
"""Get the type of the objects in the list."""
|
|
return XunitRow.__gtype__
|
|
|
|
def do_get_n_items(self) -> int:
|
|
"""Get the number of items in the list."""
|
|
return self.n_items
|
|
|
|
def do_get_item(self, n: int) -> XunitRow | None:
|
|
"""Get a specific item on the list."""
|
|
return self.__items[n] if n < self.n_items else None
|
|
|
|
def do_parse(self, rows: dict[XunitRow], row: sqlite3.Row) -> None:
|
|
"""Parse a sqlite3.Row and add it to the rows dict."""
|
|
|
|
def do_query(self, sql: sqlite.Connection) -> sqlite3.Cursor:
|
|
"""Query the database."""
|
|
return sql("SELECT name AS xunit FROM xunits WHERE runid=?",
|
|
self.runid)
|
|
|
|
def get_xunits(self) -> list[str]:
|
|
"""Get a list of xunits attached to this xfstests run."""
|
|
return sorted(self.__xunits)
|
|
|
|
|
|
class PropertyValue(XunitCell):
|
|
"""A single Property for a specific Xunit."""
|
|
|
|
key = GObject.Property(type=str)
|
|
value = GObject.Property(type=str)
|
|
|
|
def __str__(self) -> str:
|
|
"""Get a string representation of this Property."""
|
|
return f"{self.key} = {self.value}"
|
|
|
|
|
|
class Property(XunitRow):
|
|
"""Collects one property across multiple xunits."""
|
|
|
|
def do_make_xunit(self, name: str, key: str, value: str) -> PropertyValue:
|
|
"""Add a PropertyValue to the Property."""
|
|
return PropertyValue(name=name, key=key, value=value)
|
|
|
|
def all_same_value(self) -> bool:
|
|
"""Check if all the xunits have the same value."""
|
|
return len(self.get_results()) == 1
|
|
|
|
|
|
class PropertyList(XunitList):
|
|
"""A list of Properties for a specific Xfstests Run."""
|
|
|
|
def do_query(self, sql: sqlite.Connection) -> sqlite3.Cursor:
|
|
"""Query the database for properties."""
|
|
return sql("""SELECT xunit, key, value FROM xunit_properties_view
|
|
WHERE runid=?""", self.runid)
|
|
|
|
def do_parse(self, rows: dict[Property], row: sqlite3.Cursor) -> None:
|
|
"""Parse the data in the row and add it to the rows dict."""
|
|
property = rows.setdefault(row["key"], Property(row["key"]))
|
|
property.add_xunit(row["xunit"], row["key"], row["value"])
|
|
|
|
|
|
class PropertyFilter(Gtk.Filter):
|
|
"""A filter for Properties."""
|
|
|
|
def do_get_strictness(self) -> Gtk.FilterMatch:
|
|
"""Get the strictness of the filter."""
|
|
return Gtk.FilterMatch.SOME
|
|
|
|
def do_match(self, property: Property) -> bool:
|
|
"""Check if a property matches the filter."""
|
|
hidden = {"CPUS", "HOST_OPTIONS", "LOAD_FACTOR", "MEM_KB",
|
|
"NUMA_NODES", "OVL_LOWER", "OVL_UPPER", "OVL_WORK",
|
|
"PLATFORM", "SECTION", "SWAP_KB", "TIME_FACTOR"}
|
|
return property.name not in hidden
|
|
|
|
|
|
class TestResult(XunitCell):
|
|
"""The results for a single TestCase with a specific Xunit."""
|
|
|
|
status = GObject.Property(type=str)
|
|
time = GObject.Property(type=int)
|
|
message = GObject.Property(type=str)
|
|
stdout = GObject.Property(type=str)
|
|
stderr = GObject.Property(type=str)
|
|
|
|
def __str__(self) -> str:
|
|
"""Get a string representation of this TestResult."""
|
|
return self.status
|
|
|
|
|
|
class TestCase(XunitRow):
|
|
"""Collects results for a single TestCase with multiple Xunits."""
|
|
|
|
def do_make_xunit(self, name: str, status: str, time: int,
|
|
message: str | None, stdout: str | None,
|
|
stderr: str | None) -> TestResult:
|
|
"""Add an xunit result to the TestCase."""
|
|
return TestResult(name=name, status=status, time=time,
|
|
message=("" if message is None else message),
|
|
stdout=("" if stdout is None else stdout),
|
|
stderr=("" if stderr is None else stderr))
|
|
|
|
|
|
class TestCaseList(XunitList):
|
|
"""A list of TestCases for a specific Xfstests Run."""
|
|
|
|
def do_query(self, sql: sqlite.Connection) -> sqlite3.Cursor:
|
|
"""Query the database for testcase results."""
|
|
return sql("""SELECT testcase, xunit, status, time,
|
|
message, stdout, stderr
|
|
FROM testcases_view WHERE runid=?""", self.runid)
|
|
|
|
def do_parse(self, rows: dict[TestCase], row: sqlite3.Cursor) -> None:
|
|
"""Parse the data in the row and add it to the rows dict."""
|
|
testcase = rows.setdefault(row["testcase"], TestCase(row["testcase"]))
|
|
testcase.add_xunit(row["xunit"], row["status"], row["time"],
|
|
row["message"], row["stdout"], row["stderr"])
|
|
|
|
|
|
class TestCaseFilter(Gtk.Filter):
|
|
"""A filter for TestCases."""
|
|
|
|
passed = GObject.Property(type=bool, default=False)
|
|
skipped = GObject.Property(type=bool, default=False)
|
|
failure = GObject.Property(type=bool, default=True)
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize a TestCaseFilter."""
|
|
super().__init__()
|
|
self.connect("notify", self.__notify)
|
|
|
|
def __notify(self, filter: typing.Self, param: GObject.ParamSpec) -> None:
|
|
match param.name:
|
|
case "passed" | "skipped" | "failure":
|
|
if self.get_property(param.name):
|
|
change = Gtk.FilterChange.LESS_STRICT
|
|
else:
|
|
change = Gtk.FilterChange.MORE_STRICT
|
|
self.changed(change)
|
|
|
|
def do_get_strictness(self) -> Gtk.FilterMatch:
|
|
"""Get the current strictness of the filter."""
|
|
match (self.passed, self.skipped, self.failure):
|
|
case (True, True, True): return Gtk.FilterMatch.ALL
|
|
case (False, False, False): return Gtk.FilterMatch.NONE
|
|
case (_, _, _): return Gtk.FilterMatch.SOME
|
|
|
|
def do_match(self, testcase: TestCase) -> bool:
|
|
"""Check if a testcase matches the current filter."""
|
|
results = testcase.get_results()
|
|
if self.passed and "passed" in results:
|
|
return True
|
|
if self.skipped and "skipped" in results:
|
|
return True
|
|
if self.failure and "failure" in results:
|
|
return True
|
|
return False
|
|
|
|
|
|
class SummaryValue(XunitCell):
|
|
"""The summary of a single Xfstests xunit field."""
|
|
|
|
name = GObject.Property(type=str)
|
|
unit = GObject.Property(type=str)
|
|
value = GObject.Property(type=int)
|
|
|
|
def __str__(self) -> str:
|
|
"""Convert a Summary Value to a string."""
|
|
s = '' if self.value == 1 else 's'
|
|
return f"{self.value} {self.unit}{s}"
|
|
|
|
|
|
class Summary(XunitRow):
|
|
"""Collects values for each summary field with multiple Xunits."""
|
|
|
|
def __lt__(self, rhs: typing.Self) -> bool:
|
|
"""Compare the fields of two Summaries."""
|
|
order = ["passed", "failed", "skipped", "time"]
|
|
return order.index(self.name) < order.index(rhs.name)
|
|
|
|
def do_make_xunit(self, name: str, value: int, unit: str) -> SummaryValue:
|
|
"""Add an xunit summary to the Summary."""
|
|
return SummaryValue(name=name, value=value, unit=unit)
|
|
|
|
|
|
class SummaryList(XunitList):
|
|
"""A list summarizing the results of a specific Xfstests Run."""
|
|
|
|
def do_query(self, sql: sqlite.Connection) -> sqlite3.Cursor:
|
|
"""Query the database for xunit summaries."""
|
|
return sql("""SELECT name AS xunit, passed, failed, skipped, time
|
|
FROM xunits_view WHERE runid=?""", self.runid)
|
|
|
|
def do_parse(self, rows: dict[Summary], row: sqlite3.Row) -> None:
|
|
"""Parse the data in the row and add it to the rows dict."""
|
|
for field in ["passed", "failed", "skipped", "time"]:
|
|
summary = rows.setdefault(field, Summary(field))
|
|
summary.add_xunit(row["xunit"], row[field],
|
|
"second" if field == "time" else "testcase")
|