Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
79 changes: 52 additions & 27 deletions python/lib/sift_client/_tests/util/test_test_results_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,10 @@


class TestContextManager:
def test_link_run_to_report(self, report_context, nostromo_run):
report_context.report.update({"run_id": nostromo_run.id_})
assert report_context.report.run_id == nostromo_run.id_

def test_new_step(self, report_context):
initial_end_time = report_context.report.end_time
first_step_path = report_context.get_next_step_path()
Expand Down Expand Up @@ -124,41 +128,62 @@ def test_measurement_update(self, report_context):
assert test_step.measurements[2].measurement_type == TestMeasurementType.BOOLEAN

def test_bad_assert(self, report_context, step):
test_step = None
parent_step_path = step.current_step.step_path
initial_open_step_result = report_context.open_step_results.get(parent_step_path, True)
# Capture current state of report context's failures so we can keep things passed at a high level if the test's induced failures happen as expected.
current_step_path = step.current_step.step_path
initial_open_step_result = report_context.open_step_results.get(current_step_path, True)
initial_any_failures = report_context.any_failures

with step.substep("Test Bad Assert", "Test Bad Assert Description") as new_step:
test_step = new_step.current_step
assert False == True

assert test_step.status == TestStatus.ERROR
assert test_step.error_info is not None
assert "AssertionError" in test_step.error_info.error_message
parent_step = None
substep = None
nested_substep = None
sibling_substep = None
with step.substep("Top Level Step", "Should fail") as parent_step_context:
parent_step = parent_step_context.current_step
with parent_step_context.substep("Parent Step", "Should fail") as substep_context:
substep = substep_context.current_step
with substep_context.substep(
"Nested Substep", "Has a bad assert"
) as nested_substep_context:
nested_substep = nested_substep_context.current_step
assert False == True
with substep_context.substep(
"Sibling Substep", "Should pass"
) as sibling_substep_context:
sibling_substep = sibling_substep_context.current_step

assert parent_step.status == TestStatus.FAILED
assert substep.status == TestStatus.FAILED
assert nested_substep.status == TestStatus.ERROR
assert "AssertionError" in nested_substep.error_info.error_message
assert sibling_substep.status == TestStatus.PASSED

# If this test was successful, mark that at a high level.
if initial_open_step_result:
report_context.open_step_results[parent_step_path] = True
report_context.open_step_results[current_step_path] = True
if not initial_any_failures:
report_context.any_failures = False

def test_error_info(self, report_context, step):
def test_manually_skip_step(self, step):
test_step = None
parent_step_path = step.current_step.step_path
initial_open_step_result = report_context.open_step_results.get(parent_step_path, True)
initial_any_failures = report_context.any_failures
substep = None
sibling_substep = None
with step.substep("Parent Step", "Should pass") as parent_step_context:
test_step = parent_step_context.current_step
with parent_step_context.substep("Substep", "Should skip") as substep_context:
substep = substep_context.current_step
substep.update({"status": TestStatus.SKIPPED})
with substep_context.substep(
"Sibling Substep", "Should pass"
) as sibling_substep_context:
sibling_substep = sibling_substep_context.current_step

with report_context.new_step("Test Error", "Test Error Description") as new_step:
test_step = new_step.current_step
raise Exception("Test Error")
assert test_step.error_info is not None
assert test_step.error_info.error_code == 1
assert "Test Error" in test_step.error_info.error_message
assert test_step.status == TestStatus.ERROR
# If the parent step is not marked as failed already, make sure it remains passed at this point.
if initial_open_step_result:
report_context.open_step_results[parent_step_path] = True
if not initial_any_failures:
report_context.any_failures = False
assert test_step.status == TestStatus.PASSED
assert substep.status == TestStatus.SKIPPED
assert sibling_substep.status == TestStatus.PASSED

@pytest.mark.skip(reason="Test Skip Step")
def test_pytest_skip(self):
pass


class TestBounds:
Expand Down
12 changes: 12 additions & 0 deletions python/lib/sift_client/util/test_results/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,18 @@
return result # This is optional for other uses, but the step and its parents will be updated correctly i.e. failed if the measurement fails.
```

#### Manually Updating Underlyling Report
You can also manually update the underlying report or steps by accessing the context manager's attributes.
```python
with ReportContext(client, name="Example Report") as rc:
with rc.new_step(name="Example Step") as step:
if !conditions:
step.update({"status": TestStatus.SKIPPED})
else:
step.measure(name="Example Measurement", value=test_value, bounds={"min": -1, "max": 10})
rc.report.update({"run_id": run_id})
```

For a larger class or script, consider creating the context in a setup method and passing it to the test functions.
```python
def main(self):
Expand Down
14 changes: 8 additions & 6 deletions python/lib/sift_client/util/test_results/context_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,6 @@ def report_measurement(self, measurement: TestMeasurement, step: TestStep):
def resolve_and_propagate_step_result(
self,
step: TestStep,
parent_step: TestStep | None = None,
error_info: ErrorInfo | None = None,
) -> bool:
"""Resolve the result of a step and propagate the result to the parent step if it failed."""
Expand All @@ -157,13 +156,16 @@ def resolve_and_propagate_step_result(
result = False
if step.status != TestStatus.IN_PROGRESS:
# The step was manually completed so use that result.
result = step.status == TestStatus.PASSED
# Skipped steps are considered passed.
result = step.status in (TestStatus.PASSED, TestStatus.SKIPPED)

# Update the parent step results if this step failed (true by default so no need to do anything if we didn't fail).
if not result:
self.any_failures = True
if parent_step:
self.open_step_results[parent_step.step_path] = False
path_parts = step.step_path.split(".")
if len(path_parts) > 1:
parent_step_path = ".".join(path_parts[:-1])
self.open_step_results[parent_step_path] = False

return result

Expand All @@ -185,7 +187,6 @@ class NewStep(AbstractContextManager):
report_context: ReportContext
client: SiftClient
current_step: TestStep | None = None
parent_step: TestStep | None = None

def __init__(
self,
Expand Down Expand Up @@ -237,12 +238,13 @@ def update_step_from_result(

# Resolve the status of this step (i.e. fail if children failed) and propagate the result to the parent step.
result = self.report_context.resolve_and_propagate_step_result(
self.current_step, self.parent_step, error_info
self.current_step, error_info
)

# Mark the step as completed
status = self.current_step.status
if status == TestStatus.IN_PROGRESS:
# Update the status only if the step was in progress i.e. not updated elsewhere.
status = TestStatus.PASSED if result else TestStatus.FAILED
if error_info:
status = TestStatus.ERROR
Expand Down
15 changes: 13 additions & 2 deletions python/lib/sift_client/util/test_results/pytest_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,19 +6,27 @@

import pytest

from sift_client.sift_types.test_report import TestStatus
from sift_client.util.test_results import ReportContext

if TYPE_CHECKING:
from sift_client.client import SiftClient
from sift_client.util.test_results.context_manager import NewStep

REPORT_CONTEXT: ReportContext | None = None


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[Any]):
"""You should import this hook to capture any AssertionErrors that occur during the test. If not included, any assert failures in a test will not automatically fail the step."""
outcome = yield
rep = outcome.get_result()
setattr(item, "rep_" + rep.when, call)
report = outcome.get_result()
if report.outcome == "skipped":
# Skipped steps won't invoke the method/fixtures at all, so we need to manually record a step.
if REPORT_CONTEXT:
with REPORT_CONTEXT.new_step(name=item.name) as new_step:
new_step.current_step.update({"status": TestStatus.SKIPPED})
setattr(item, "rep_" + report.when, call)


def _report_context_impl(
Expand All @@ -36,6 +44,9 @@ def _report_context_impl(
name=f"{base_name} {datetime.now(timezone.utc).isoformat()}",
test_case=str(test_case),
) as context:
# Set a global so we can access this in pytest hooks.
global REPORT_CONTEXT
REPORT_CONTEXT = context
yield context


Expand Down
Loading