[uss_qualifier/reports] Typing fixes

Author: the-glu

.basedpyright/baseline.json

@@ -110,6 +110,7 @@ def generate_globally_expanded_report(
 
 """
 
+    assert report.configuration.v1 and report.configuration.v1.test_run
     resource_pool = make_resources_config(report.configuration.v1.test_run)
 
     def indented_ul(value) -> list[str]:
@@ -184,6 +185,7 @@ def describe_pool_resource(k: str, v: dict) -> str:
 
 def _generate_sections(node: ActionNode) -> Iterator[_Section]:
     if node.node_type == ActionNodeType.Scenario:
+        assert node.scenario
         yield _generate_scenario_section(node.scenario)
     elif node.node_type == ActionNodeType.SkippedAction:
         yield _generate_skipped_scenario_section(node)
@@ -195,7 +197,7 @@ def _generate_sections(node: ActionNode) -> Iterator[_Section]:
 def _generate_skipped_scenario_section(node: ActionNode) -> _Section:
     return _Section(
         title=f"[skipped] {node.name}",
-        body=f"This instance of this test scenario was skipped in this test run because: {node.skipped_action.reason}",
+        body=f"This instance of this test scenario was skipped in this test run because: {node.skipped_action.reason if node.skipped_action else '?'}",
     )
 
 
@@ -237,15 +239,15 @@ def _indent_headings(elements: Sequence[marko.element.Element], levels: int) ->
     for element in elements:
         if isinstance(element, marko.block.Heading):
             element.level = min(element.level + levels, 6)
-        if hasattr(element, "children") and element.children:
-            _indent_headings(element.children, levels)
+        if children := getattr(element, "children", None):
+            _indent_headings(children, levels)
 
 
 def _inflate_fragments(parent: marko.element.Element, origin_filename: str) -> None:
-    if hasattr(parent, "children") and parent.children:
+    if hasattr(parent, "children") and parent.children:  # pyright: ignore[reportAttributeAccessIssue]
         c = 0
-        while c < len(parent.children):
-            child = parent.children[c]
+        while c < len(parent.children):  # pyright: ignore[reportAttributeAccessIssue]
+            child = parent.children[c]  # pyright: ignore[reportAttributeAccessIssue]
             if (
                 isinstance(child, marko.block.Heading)
                 and hasattr(child, "children")
@@ -260,12 +262,12 @@ def _inflate_fragments(parent: marko.element.Element, origin_filename: str) -> N
                 doc = _get_test_step_fragment(absolute_path, child.level)
                 _update_links(doc, absolute_path)
                 _strip_link(child)
-                parent.children = (
-                    parent.children[0 : c + 1] + doc.children + parent.children[c + 1 :]
+                parent.children = (  # pyright: ignore[reportAttributeAccessIssue]
+                    parent.children[0 : c + 1] + doc.children + parent.children[c + 1 :]  # pyright: ignore[reportAttributeAccessIssue]
                 )
                 c += len(doc.children)
             elif isinstance(child, marko.element.Element):
-                _inflate_fragments(parent.children[c], origin_filename)
+                _inflate_fragments(parent.children[c], origin_filename)  # pyright: ignore[reportAttributeAccessIssue]
                 c += 1
             else:
                 c += 1
@@ -286,14 +288,14 @@ def add_resource_origin():
             note = marko.parse(
                 """&empty; _This resource was not applicable to this test run and was therefore not provided._\n\n"""
             )
-            doc.children = doc.children[0:c] + note.children + doc.children[c:]
+            doc.children = doc.children[0:c] + note.children + doc.children[c:]  # pyright: ignore[reportAttributeAccessIssue,reportOperatorIssue]
             c += len(note.children)
             return
         # Insert resource origin information
         origin = marko.parse(
             f"\n\n&#x2705; Provided by {scenario.resource_origins[current_resource]}.\n"
         )
-        doc.children = doc.children[0:c] + origin.children + doc.children[c:]
+        doc.children = doc.children[0:c] + origin.children + doc.children[c:]  # pyright: ignore[reportOperatorIssue]
         c += len(origin.children)
 
     while c < len(doc.children):
@@ -327,11 +329,11 @@ def add_resource_origin():
 
 
 def _strip_link(element: marko.element.Element) -> None:
-    if hasattr(element, "children") and element.children:
-        for c in range(len(element.children)):
-            child = element.children[c]
+    if hasattr(element, "children") and element.children:  # pyright: ignore[reportAttributeAccessIssue]
+        for c in range(len(element.children)):  # pyright: ignore[reportAttributeAccessIssue]
+            child = element.children[c]  # pyright: ignore[reportAttributeAccessIssue]
             if isinstance(child, marko.block.inline.Link):
-                element.children[c] = child.children[0]
+                element.children[c] = child.children[0]  # pyright: ignore[reportAttributeAccessIssue]
             elif isinstance(child, marko.element.Element):
                 _strip_link(child)
 
@@ -372,7 +374,7 @@ def add_context_to_case():
                     """&empty; _This test case was not applicable to this test run and is therefore not statused._\n\n"""
                 )
                 doc.children = (
-                    doc.children[0 : test_case_i0 + 1]
+                    doc.children[0 : test_case_i0 + 1]  # pyright: ignore[reportAttributeAccessIssue,reportOperatorIssue]
                     + note.children
                     + doc.children[test_case_i0 + 1 :]
                 )
@@ -396,6 +398,7 @@ def add_context_to_case():
             for epoch in scenario.epochs:
                 if epoch.type != EpochType.Case:
                     continue
+                assert epoch.case
                 if case_name == epoch.case.name:
                     test_case = epoch.case
                     break
@@ -409,6 +412,7 @@ def add_context_to_case():
             for epoch in scenario.epochs:
                 if epoch.type != EpochType.Case:
                     continue
+                assert epoch.case
                 if len(epoch.case.steps) == 1 and epoch.case.steps[0].name == "Cleanup":
                     test_case = epoch.case
                     break
@@ -444,7 +448,7 @@ def add_context_to_step():
                 )
                 dc = len(note.children)
                 doc.children = (
-                    doc.children[0 : test_step_i0 + 1]
+                    doc.children[0 : test_step_i0 + 1]  # pyright: ignore[reportOperatorIssue]
                     + note.children
                     + doc.children[test_step_i0 + 1 :]
                 )
@@ -494,6 +498,7 @@ def _add_context_to_step(
     def add_context_to_check():
         nonlocal c, i1, added, test_check_name, test_check_i0, test_check_level
         if test_check_name is not None:
+            assert test_check_i0 is not None
             dc = _add_context_to_check(doc, step, test_check_name, test_check_i0, c)
             c += dc
             i1 += dc
@@ -534,13 +539,15 @@ def _add_context_to_check(
     for event in step.events:
         if (
             event.type == EventType.PassedCheck
+            and event.passed_check is not None
             and event.passed_check.name == test_check_name
         ):
             check_text.append(
                 f"&#x2705; {', '.join(event.passed_check.participants)} ({event.passed_check.timestamp})"
             )
         elif (
             event.type == EventType.FailedCheck
+            and event.failed_check
             and event.failed_check.name == test_check_name
         ):
             check_text.append(
@@ -552,7 +559,7 @@ def _add_context_to_check(
         additions = marko.parse(
             """&empty; _This check was not applicable to this test run and is therefore not statused._\n\n"""
         )
-    doc.children = doc.children[0:i1] + additions.children + doc.children[i1:]
+    doc.children = doc.children[0:i1] + additions.children + doc.children[i1:]  # pyright: ignore[reportOperatorIssue]
     return len(additions.children)
 
 
@@ -571,16 +578,16 @@ def _update_links(element: marko.element.Element, origin_filename: str) -> None:
                 url = url.replace("/github.com/", "/raw.githubusercontent.com/")
                 url = url.replace("/blob/", "/")
             element.dest = url
-    if hasattr(element, "children") and element.children:
-        for child in element.children:
+    if hasattr(element, "children") and element.children:  # pyright: ignore[reportAttributeAccessIssue]
+        for child in element.children:  # pyright: ignore[reportAttributeAccessIssue]
             if isinstance(child, marko.element.Element):
                 _update_links(child, origin_filename)
 
 
 def _add_section_numbers(elements: Sequence[marko.element.Element]) -> None:
     heading_level = 2
     levels = [0]
-    headings = [None]
+    headings: list[str | None] = [None]
     prev_heading = None
     for i, element in enumerate(elements):
         if isinstance(element, marko.block.Heading):
@@ -599,7 +606,7 @@ def _add_section_numbers(elements: Sequence[marko.element.Element]) -> None:
                 heading_level += 1
             else:
                 headings.append(text_of(element))
-                heading_trace = " -> ".join(headings)
+                heading_trace = " -> ".join([str(heading) for heading in headings])
                 raise ValueError(
                     f"Encountered a level {element.level} heading ({text_of(element)}) at element {i} following a level {heading_level} heading ({prev_heading}); expected heading levels to increase by 1 level at a time.  Trace: {heading_trace}"
                 )
@@ -612,4 +619,4 @@ def _add_section_numbers(elements: Sequence[marko.element.Element]) -> None:
             else:
                 element.children = [
                     marko.block.inline.RawText(section_number)
-                ] + element.children
+                ] + element.children  # pyright: ignore[reportOperatorIssue]

monitoring/uss_qualifier/reports/globally_expanded/generate.py

@@ -357,12 +357,13 @@ def queries(self) -> list[fetch.Query]:
         for case in self.cases:
             for step in case.steps:
                 if step.has_field_with_value("queries"):
+                    assert step.queries
                     queries.extend(step.queries)
 
-        if self.has_field_with_value("cleanup") and self.cleanup.has_field_with_value(
+        if self.has_field_with_value("cleanup") and self.cleanup.has_field_with_value(  # pyright: ignore[reportOptionalMemberAccess]
             "queries"
         ):
-            queries.extend(self.cleanup.queries)
+            queries.extend(self.cleanup.queries)  # pyright: ignore[reportOptionalMemberAccess,reportArgumentType]
 
         return queries
 

monitoring/uss_qualifier/reports/report.py

@@ -67,7 +67,7 @@ def _step_events(
     scenario_participants: dict[ParticipantID, TestedParticipant],
     all_events: list[Event],
     after: datetime | None,
-) -> tuple[TestedStep, datetime]:
+) -> tuple[TestedStep, datetime | None]:
     events = []
 
     # Create events for this step's passed checks
@@ -117,6 +117,7 @@ def _step_events(
                 for e in all_events:
                     if (
                         e.type == EventType.Query
+                        and e.query is not None
                         and e.query.request.initiated_at == query_timestamp
                     ):
                         query_events.append(e)

monitoring/uss_qualifier/reports/sequence_view/events.py

@@ -51,6 +51,7 @@
 
 def _skipped_action_of(report: SkippedActionReport) -> ActionNode:
     if report.declaration.get_action_type() == ActionType.TestSuite:
+        assert report.declaration.test_suite
         if (
             "suite_type" in report.declaration.test_suite
             and report.declaration.test_suite.suite_type
@@ -76,6 +77,7 @@ def _skipped_action_of(report: SkippedActionReport) -> ActionNode:
             )
         name = "All actions in test suite"
     elif report.declaration.get_action_type() == ActionType.TestScenario:
+        assert report.declaration.test_scenario
         docs = get_documentation_by_name(report.declaration.test_scenario.scenario_type)
         return ActionNode(
             name=docs.name,
@@ -84,6 +86,7 @@ def _skipped_action_of(report: SkippedActionReport) -> ActionNode:
             skipped_action=SkippedAction(reason=report.reason),
         )
     elif report.declaration.get_action_type() == ActionType.ActionGenerator:
+        assert report.declaration.action_generator
         generator_type = action_generator_type_from_name(
             report.declaration.action_generator.generator_type
         )
@@ -123,20 +126,23 @@ def compute_action_node(report: TestSuiteActionReport, indexer: Indexer) -> Acti
         is_action_generator,
     ) = report.get_applicable_report()
     if is_test_scenario:
+        assert report.test_scenario
         return ActionNode(
             name=report.test_scenario.name,
             node_type=ActionNodeType.Scenario,
             children=[],
             scenario=compute_tested_scenario(report.test_scenario, indexer),
         )
     elif is_test_suite:
+        assert report.test_suite
         children = [compute_action_node(a, indexer) for a in report.test_suite.actions]
         return ActionNode(
             name=report.test_suite.name,
             node_type=ActionNodeType.Suite,
             children=children,
         )
     elif is_action_generator:
+        assert report.action_generator
         generator_type = action_generator_type_from_name(
             report.action_generator.generator_type
         )
@@ -148,6 +154,7 @@ def compute_action_node(report: TestSuiteActionReport, indexer: Indexer) -> Acti
             ],
         )
     else:
+        assert report.skipped_action
         return _skipped_action_of(report.skipped_action)
 
 
@@ -177,9 +184,13 @@ def _align_overview_rows(rows: list[OverviewRow]) -> None:
             row.filled = True
             to_fill -= 1
         elif len(row.suite_cells) < max_suite_cols:
-            if row.suite_cells[-1].first_row and all(
-                c.node_type == ActionNodeType.Scenario
-                for c in row.suite_cells[-1].node.children
+            if (
+                row.suite_cells[-1].first_row
+                and row.suite_cells[-1].node is not None
+                and all(
+                    c.node_type == ActionNodeType.Scenario
+                    for c in row.suite_cells[-1].node.children
+                )
             ):
                 row.suite_cells[-1].colspan += max_suite_cols - len(row.suite_cells)
                 row.filled = True
@@ -212,6 +223,7 @@ def _align_overview_rows(rows: list[OverviewRow]) -> None:
 
 def _enumerate_all_participants(node: ActionNode) -> list[ParticipantID]:
     if node.node_type == ActionNodeType.Scenario:
+        assert node.scenario
         return list(node.scenario.participants)
     else:
         result = set()
@@ -225,6 +237,7 @@ def _generate_scenario_pages(
     node: ActionNode, config: SequenceViewConfiguration, output_path: str
 ) -> None:
     if node.node_type == ActionNodeType.Scenario:
+        assert node.scenario
         all_participants = list(node.scenario.participants)
         all_participants.sort()
         if UNATTRIBUTED_PARTICIPANT in all_participants:
@@ -308,6 +321,7 @@ def generate_sequence_view(
 ) -> None:
     node = compute_action_node(report.report, Indexer())
 
+    assert report.configuration.v1 and report.configuration.v1.test_run
     resources_config = make_resources_config(report.configuration.v1.test_run)
 
     os.makedirs(output_path, exist_ok=True)