[uss_qualifier/reports] Typing fixes

Author: the-glu

.basedpyright/baseline.json

@@ -21,8 +21,6 @@
 from monitoring.uss_qualifier.reports.sequence_view.summary_types import (
     ActionNode,
     ActionNodeType,
-    EpochType,
-    EventType,
     Indexer,
     TestedCase,
     TestedScenario,
@@ -110,6 +108,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 +183,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 +195,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 +237,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 hasattr(element, "children") and element.children:  # pyright: ignore[reportAttributeAccessIssue]
+            _indent_headings(element.children, levels)  # pyright: ignore[reportAttributeAccessIssue]
 
 
 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 +260,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 +286,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 +327,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 +372,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 :]
                 )
@@ -394,7 +394,7 @@ def add_context_to_case():
             test_case_i0 = c
             test_case_level = child.level
             for epoch in scenario.epochs:
-                if epoch.type != EpochType.Case:
+                if epoch.case is None:
                     continue
                 if case_name == epoch.case.name:
                     test_case = epoch.case
@@ -407,7 +407,7 @@ def add_context_to_case():
             test_case_level = child.level
             cleanup = True
             for epoch in scenario.epochs:
-                if epoch.type != EpochType.Case:
+                if epoch.case is None:
                     continue
                 if len(epoch.case.steps) == 1 and epoch.case.steps[0].name == "Cleanup":
                     test_case = epoch.case
@@ -444,7 +444,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 +494,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
@@ -533,14 +534,14 @@ def _add_context_to_check(
     check_text = [""]
     for event in step.events:
         if (
-            event.type == EventType.PassedCheck
+            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
+            event.failed_check is not None
             and event.failed_check.name == test_check_name
         ):
             check_text.append(
@@ -552,7 +553,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 +572,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 +600,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 +613,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

@@ -15,7 +15,6 @@
 from monitoring.uss_qualifier.reports.sequence_view.summary_types import (
     Epoch,
     Event,
-    EventType,
     Indexer,
     NoteEvent,
     TestedCase,
@@ -67,7 +66,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
@@ -116,7 +115,7 @@ def _step_events(
                 found = False
                 for e in all_events:
                     if (
-                        e.type == EventType.Query
+                        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

@@ -31,8 +31,6 @@
 from monitoring.uss_qualifier.reports.sequence_view.summary_types import (
     ActionNode,
     ActionNodeType,
-    EpochType,
-    EventType,
     Indexer,
     OverviewRow,
     SkippedAction,
@@ -51,6 +49,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 +75,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 +84,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 +124,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 +152,7 @@ def compute_action_node(report: TestSuiteActionReport, indexer: Indexer) -> Acti
             ],
         )
     else:
+        assert report.skipped_action
         return _skipped_action_of(report.skipped_action)
 
 
@@ -177,9 +182,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 +221,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 +235,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:
@@ -241,8 +252,6 @@ def _generate_scenario_pages(
                     test_scenario=node.scenario,
                     all_participants=all_participants,
                     kml_file=kml_file if config.render_kml else None,
-                    EpochType=EpochType,
-                    EventType=EventType,
                     UNATTRIBUTED_PARTICIPANT=UNATTRIBUTED_PARTICIPANT,
                     len=len,
                     str=str,
@@ -308,6 +317,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)