Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(HardwareDetails): add platform card to tabs #710

Merged
merged 3 commits into from
Dec 24, 2024
Merged

feat(HardwareDetails): add platform card to tabs #710

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5794579
feat(hardware): add platform support to backend
murilx Dec 24, 2024
7063298
feat(hardware): add platform card to tabs
murilx Dec 24, 2024
92abc06
refactor: improve code formatting in filters.py
murilx Dec 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
203 changes: 110 additions & 93 deletions backend/kernelCI_app/helpers/filters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,7 @@
from typing import Optional, Dict, List
from typing import Optional, Dict, List, TypedDict, Literal, Any
from django.http import HttpResponseBadRequest
import re
from kernelCI_app.utils import (
getErrorResponseBody
)
from kernelCI_app.utils import getErrorResponseBody

UNKNOWN_STRING = "Unknown"
NULL_STRINGS = set(["null", UNKNOWN_STRING, "NULL"])
Expand Down Expand Up @@ -96,7 +94,12 @@ class FilterParams:

string_like_filters = ["boot.path", "test.path"]

def __init__(self, data: Dict, process_body=False):
class ParsedFilter(TypedDict):
field: str
value: Any # TODO: correctly type this field
comparison_op: Literal["exact", "in", "gt", "gte", "lt", "lte", "like"]

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nice

def __init__(self, data: Dict, process_body=False) -> None:
self.filterTestDurationMin, self.filterTestDurationMax = None, None
self.filterBootDurationMin, self.filterBootDurationMax = None, None
self.filterBuildDurationMin, self.filterBuildDurationMax = None, None
Expand All @@ -109,10 +112,11 @@ def __init__(self, data: Dict, process_body=False):
self.filterTestPath = ""
self.filterBootPath = ""
self.filterBuildValid = set()
self.filterIssues = {
self.filterIssues = {"build": set(), "boot": set(), "test": set()}
self.filterPlatforms = {
"build": set(),
"boot": set(),
"test": set()
"test": set(),
}

self.filter_handlers = {
Expand All @@ -131,71 +135,78 @@ def __init__(self, data: Dict, process_body=False):
"build.issue": self._handle_issues,
"boot.issue": self._handle_issues,
"test.issue": self._handle_issues,
"build.platform": self._handle_platforms,
"boot.platform": self._handle_platforms,
"test.platform": self._handle_platforms,
}

self.filters = []
self.filters: List[FilterParams.ParsedFilter] = []
if process_body:
self.create_filters_from_body(data)
else:
self.create_filters_from_req(data)

self._processFilters()

def _handle_boot_status(self, current_filter: Dict):
def _handle_boot_status(self, current_filter: ParsedFilter) -> None:
self.filterBootStatus.add(current_filter["value"])

def _handle_boot_duration(self, current_filter: Dict):
def _handle_boot_duration(self, current_filter: ParsedFilter) -> None:
value = current_filter["value"]
operation = current_filter["comparison_op"]
if operation == "lte":
self.filterBootDurationMax = toIntOrDefault(value, None)
else:
self.filterBootDurationMin = toIntOrDefault(value, None)

def _handle_test_status(self, current_filter: Dict):
def _handle_test_status(self, current_filter: ParsedFilter) -> None:
self.filterTestStatus.add(current_filter["value"])

def _handle_test_duration(self, current_filter: Dict):
def _handle_test_duration(self, current_filter: ParsedFilter) -> None:
value = current_filter["value"]
operation = current_filter["comparison_op"]
if operation == "lte":
self.filterTestDurationMax = toIntOrDefault(value, None)
else:
self.filterTestDurationMin = toIntOrDefault(value, None)

def _handle_config_name(self, current_filter: Dict):
def _handle_config_name(self, current_filter: ParsedFilter) -> None:
self.filterConfigs.add(current_filter["value"])

def _handle_compiler(self, current_filter: Dict):
def _handle_compiler(self, current_filter: ParsedFilter) -> None:
self.filterCompiler.add(current_filter["value"])

def _handle_architecture(self, current_filter: Dict):
def _handle_architecture(self, current_filter: ParsedFilter) -> None:
self.filterArchitecture.add(current_filter["value"])

def _handle_hardware(self, current_filter: Dict):
def _handle_hardware(self, current_filter: ParsedFilter) -> None:
self.filterHardware.add(current_filter["value"])

def _handle_path(self, current_filter: Dict):
def _handle_path(self, current_filter: ParsedFilter) -> None:
if current_filter["field"] == "boot.path":
self.filterBootPath = current_filter["value"]
else:
self.filterTestPath = current_filter["value"]

def _handle_build_valid(self, current_filter: Dict):
def _handle_build_valid(self, current_filter: ParsedFilter) -> None:
self.filterBuildValid.add(current_filter["value"])

def _handle_build_duration(self, current_filter: Dict):
def _handle_build_duration(self, current_filter: ParsedFilter) -> None:
value = current_filter["value"][0]
operation = current_filter["comparison_op"]
if operation == "lte":
self.filterBuildDurationMax = toIntOrDefault(value, None)
else:
self.filterBuildDurationMin = toIntOrDefault(value, None)

def _handle_issues(self, current_filter):
tab = current_filter["field"].split('.')[0]
def _handle_issues(self, current_filter: ParsedFilter) -> None:
tab = current_filter["field"].split(".")[0]
self.filterIssues[tab].add(current_filter["value"])

def _handle_platforms(self, current_filter: ParsedFilter) -> None:
tab = current_filter["field"].split(".")[0]
self.filterPlatforms[tab].add(current_filter["value"])

def _processFilters(self):
try:
for current_filter in self.filters:
Expand Down Expand Up @@ -270,7 +281,7 @@ def create_filters_from_req(self, request):

self.add_filter(filter_term, request.GET.get(k), "exact")

def add_filter(self, field, value, comparison_op):
def add_filter(self, field: str, value: Any, comparison_op: str) -> None:
self.validate_comparison_op(comparison_op)
self.filters.append(
{"field": field, "value": value, "comparison_op": comparison_op}
Expand All @@ -291,50 +302,64 @@ def get_grouped_filters(self):

for f in self.filters:
field = f["field"]
value = f['value']
value = f["value"]
if field not in grouped_filters:
grouped_filters[field] = f
elif type(grouped_filters[field]['value']) is str:
grouped_filters[field]['value'] = [grouped_filters[field]['value'], value]
elif type(grouped_filters[field]["value"]) is str:
grouped_filters[field]["value"] = [
grouped_filters[field]["value"],
value,
]
else:
grouped_filters[field]['value'].append(value)
grouped_filters[field]["value"].append(value)

return grouped_filters

def is_build_filtered_out(
self, *,
duration: Optional[int],
valid: Optional[bool],
issue_id: Optional[str]
self,
*,
duration: Optional[int],
valid: Optional[bool],
issue_id: Optional[str],
platform: Optional[str] = None,
) -> bool:
return (
len(self.filterBuildValid) > 0
and (str(valid).lower() not in self.filterBuildValid)
) or (
(self.filterBuildDurationMax is not None or self.filterBuildDurationMin is not None)
and duration is None
) or (
self.filterBuildDurationMax is not None and (
toIntOrDefault(duration, 0) > self.filterBuildDurationMax
(
len(self.filterBuildValid) > 0
and (str(valid).lower() not in self.filterBuildValid)
)
) or (
self.filterBuildDurationMin is not None and (
toIntOrDefault(duration, 0) < self.filterBuildDurationMin
or (
(
self.filterBuildDurationMax is not None
or self.filterBuildDurationMin is not None
)
and duration is None
)
or (
self.filterBuildDurationMax is not None
and (toIntOrDefault(duration, 0) > self.filterBuildDurationMax)
)
or (
self.filterBuildDurationMin is not None
and (toIntOrDefault(duration, 0) < self.filterBuildDurationMin)
)
or (
len(self.filterIssues["build"]) > 0
and (issue_id not in self.filterIssues["build"] or valid is True)
)
) or (
len(self.filterIssues["build"]) > 0
and (
issue_id not in self.filterIssues["build"]
or valid is True
or (
len(self.filterPlatforms["build"]) > 0
and (platform not in self.filterPlatforms["build"])
)
)

def is_record_filtered_out(
self, *,
hardwares: Optional[List[str]] = None,
architecture: Optional[str],
compiler: Optional[str],
config_name: Optional[str]
self,
*,
hardwares: Optional[List[str]] = None,
architecture: Optional[str],
compiler: Optional[str],
config_name: Optional[str],
) -> bool:
hardware_compatibles = [UNKNOWN_STRING]
record_architecture = UNKNOWN_STRING
Expand Down Expand Up @@ -373,21 +398,19 @@ def is_record_filtered_out(
return False

def is_boot_filtered_out(
self, *,
path: Optional[str],
status: Optional[str],
duration: Optional[int],
issue_id: Optional[str] = None,
incident_test_id: Optional[str] = "incident_test_id",
self,
*,
path: Optional[str],
status: Optional[str],
duration: Optional[int],
issue_id: Optional[str] = None,
incident_test_id: Optional[str] = "incident_test_id",
platform: Optional[str] = None,
) -> bool:
if (
(
self.filterBootPath != ""
and (self.filterBootPath not in path)
)
(self.filterBootPath != "" and (self.filterBootPath not in path))
or (
len(self.filterBootStatus) > 0
and (status not in self.filterBootStatus)
len(self.filterBootStatus) > 0 and (status not in self.filterBootStatus)
)
or (
(
Expand All @@ -398,42 +421,38 @@ def is_boot_filtered_out(
)
or (
self.filterBootDurationMax is not None
and (
toIntOrDefault(duration, 0) > self.filterBootDurationMax
)
and (toIntOrDefault(duration, 0) > self.filterBootDurationMax)
)
or (
self.filterBootDurationMin is not None
and (
toIntOrDefault(duration, 0) < self.filterBootDurationMin
)
and (toIntOrDefault(duration, 0) < self.filterBootDurationMin)
)
or should_filter_test_issue(
self.filterIssues["boot"],
issue_id,
incident_test_id
self.filterIssues["boot"], issue_id, incident_test_id
)
or (
len(self.filterPlatforms["boot"]) > 0
and (platform not in self.filterPlatforms["boot"])
)
):
return True

return False

def is_test_filtered_out(
self, *,
path: Optional[str],
status: Optional[str],
duration: Optional[int],
issue_id: Optional[str] = None,
incident_test_id: Optional[str] = "incident_test_id",
self,
*,
path: Optional[str],
status: Optional[str],
duration: Optional[int],
issue_id: Optional[str] = None,
incident_test_id: Optional[str] = "incident_test_id",
platform: Optional[str] = None,
) -> bool:
if (
(
self.filterTestPath != ""
and (self.filterTestPath not in path)
)
(self.filterTestPath != "" and (self.filterTestPath not in path))
or (
len(self.filterTestStatus) > 0
and (status not in self.filterTestStatus)
len(self.filterTestStatus) > 0 and (status not in self.filterTestStatus)
)
or (
(
Expand All @@ -444,20 +463,18 @@ def is_test_filtered_out(
)
or (
self.filterTestDurationMax is not None
and (
toIntOrDefault(duration, 0) > self.filterTestDurationMax
)
and (toIntOrDefault(duration, 0) > self.filterTestDurationMax)
)
or (
self.filterTestDurationMin is not None
and (
toIntOrDefault(duration, 0) < self.filterTestDurationMin
)
and (toIntOrDefault(duration, 0) < self.filterTestDurationMin)
)
or should_filter_test_issue(
self.filterIssues["test"],
issue_id,
incident_test_id
self.filterIssues["test"], issue_id, incident_test_id
)
or (
len(self.filterPlatforms["test"]) > 0
and (platform not in self.filterPlatforms["test"])
)
):
return True
Expand Down
5 changes: 2 additions & 3 deletions backend/kernelCI_app/viewCommon.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from typing import TypedDict
from kernelCI_app.utils import create_issue
from kernelCI_app.helpers.build import build_status_map


class BuildDict(TypedDict):
Expand All @@ -14,14 +15,12 @@ def create_default_build_status():


def create_details_build_summary(builds: list[BuildDict]):
status_map = {True: "valid", False: "invalid", None: "null"}

build_summ = create_default_build_status()
config_summ = {}
arch_summ = {}

for build in builds:
status_key = status_map[build["valid"]]
status_key = build_status_map[build["valid"]]
build_summ[status_key] += 1

if config := build["config_name"]:
Expand Down
Loading
Loading