9
9
import json
10
10
import os
11
11
from pathlib import Path
12
- from typing import Optional
12
+ from typing import Dict , Optional
13
13
14
- import pluggy
15
14
import pytest
15
+ from pluggy import Result
16
+ from pytest import StashKey , TestReport
16
17
17
18
from localstack .testing .aws .util import is_aws_cloud
18
19
20
+ durations_key = StashKey [Dict [str , float ]]()
21
+ """
22
+ Stores phase durations on the test node between execution phases.
23
+ See https://docs.pytest.org/en/latest/reference/reference.html#pytest.Stash
24
+ """
25
+ test_failed_key = StashKey [bool ]()
26
+ """
27
+ Stores information from call execution phase about whether the test failed.
28
+ """
29
+
19
30
20
- def find_snapshot_for_item (item : pytest .Item ) -> Optional [dict ]:
31
+ def find_validation_data_for_item (item : pytest .Item ) -> Optional [dict ]:
21
32
base_path = os .path .join (item .fspath .dirname , item .fspath .purebasename )
22
- snapshot_path = f"{ base_path } .snapshot .json"
33
+ snapshot_path = f"{ base_path } .validation .json"
23
34
24
35
if not os .path .exists (snapshot_path ):
25
36
return None
@@ -29,65 +40,76 @@ def find_snapshot_for_item(item: pytest.Item) -> Optional[dict]:
29
40
return file_content .get (item .nodeid )
30
41
31
42
32
- def find_validation_data_for_item (item : pytest .Item ) -> Optional [dict ]:
33
- base_path = os .path .join (item .fspath .dirname , item .fspath .purebasename )
34
- snapshot_path = f"{ base_path } .validation.json"
43
+ @pytest .hookimpl (hookwrapper = True )
44
+ def pytest_runtest_makereport (item : pytest .Item , call : pytest .CallInfo ):
45
+ """
46
+ This hook is called after each test execution phase (setup, call, teardown).
47
+ """
48
+ result : Result = yield
49
+ report : TestReport = result .get_result ()
35
50
36
- if not os .path .exists (snapshot_path ):
37
- return None
51
+ if call .when == "setup" :
52
+ _makereport_setup (item , call )
53
+ elif call .when == "call" :
54
+ _makereport_call (item , call )
55
+ elif call .when == "teardown" :
56
+ _makereport_teardown (item , call )
38
57
39
- with open (snapshot_path , "r" ) as fd :
40
- file_content = json .load (fd )
41
- return file_content .get (item .nodeid )
58
+ return report
42
59
43
60
44
- def record_passed_validation (item : pytest .Item , timestamp : Optional [datetime .datetime ] = None ):
61
+ def _stash_phase_duration (call , item ):
62
+ durations_by_phase = item .stash .setdefault (durations_key , {})
63
+ durations_by_phase [call .when ] = round (call .duration , 2 )
64
+
65
+
66
+ def _makereport_setup (item : pytest .Item , call : pytest .CallInfo ):
67
+ _stash_phase_duration (call , item )
68
+
69
+
70
+ def _makereport_call (item : pytest .Item , call : pytest .CallInfo ):
71
+ _stash_phase_duration (call , item )
72
+ item .stash [test_failed_key ] = call .excinfo is not None
73
+
74
+
75
+ def _makereport_teardown (item : pytest .Item , call : pytest .CallInfo ):
76
+ _stash_phase_duration (call , item )
77
+
78
+ # only update the file when running against AWS and the test finishes successfully
79
+ if not is_aws_cloud () or item .stash .get (test_failed_key , True ):
80
+ return
81
+
45
82
base_path = os .path .join (item .fspath .dirname , item .fspath .purebasename )
46
83
file_path = Path (f"{ base_path } .validation.json" )
47
84
file_path .touch ()
48
85
with file_path .open (mode = "r+" ) as fd :
49
86
# read existing state from file
50
87
try :
51
88
content = json .load (fd )
52
- except json .JSONDecodeError : # expected on first try (empty file)
89
+ except json .JSONDecodeError : # expected on the first try (empty file)
53
90
content = {}
54
91
55
- # update for this pytest node
56
- if not timestamp :
57
- timestamp = datetime .datetime .now (tz = datetime .timezone .utc )
58
- content [item .nodeid ] = {"last_validated_date" : timestamp .isoformat (timespec = "seconds" )}
92
+ test_execution_data = content .setdefault (item .nodeid , {})
59
93
60
- # save updates
61
- fd .seek (0 )
62
- json .dump (content , fd , indent = 2 , sort_keys = True )
63
- fd .write ("\n " ) # add trailing newline for linter and Git compliance
94
+ timestamp = datetime .datetime .now (tz = datetime .timezone .utc )
95
+ test_execution_data ["last_validated_date" ] = timestamp .isoformat (timespec = "seconds" )
64
96
97
+ durations_by_phase = item .stash [durations_key ]
98
+ test_execution_data ["durations_in_seconds" ] = durations_by_phase
65
99
66
- # TODO: we should skip if we're updating snapshots
67
- # make sure this is *AFTER* snapshot comparison => tryfirst=True
68
- @pytest .hookimpl (hookwrapper = True , tryfirst = True )
69
- def pytest_runtest_call (item : pytest .Item ):
70
- outcome : pluggy .Result = yield
100
+ total_duration = sum (durations_by_phase .values ())
101
+ durations_by_phase ["total" ] = round (total_duration , 2 )
71
102
72
- # we only want to track passed runs against AWS
73
- if not is_aws_cloud () or outcome .excinfo :
74
- return
103
+ # For json.dump sorted test entries enable consistent diffs.
104
+ # But test execution data is more readable in insert order for each step (setup, call, teardown).
105
+ # Hence, not using global sort_keys=True for json.dump but rather additionally sorting top-level dict only.
106
+ content = dict (sorted (content .items ()))
75
107
76
- record_passed_validation (item )
77
-
78
-
79
- # this is a sort of utility used for retroactively creating validation files in accordance with existing snapshot files
80
- # it takes the recorded date from a snapshot and sets it to the last validated date
81
- # @pytest.hookimpl(trylast=True)
82
- # def pytest_collection_modifyitems(session: pytest.Session, config: pytest.Config, items: list[pytest.Item]):
83
- # for item in items:
84
- # snapshot_entry = find_snapshot_for_item(item)
85
- # if not snapshot_entry:
86
- # continue
87
- #
88
- # snapshot_update_timestamp = datetime.datetime.strptime(snapshot_entry["recorded-date"], "%d-%m-%Y, %H:%M:%S").astimezone(tz=datetime.timezone.utc)
89
- #
90
- # record_passed_validation(item, snapshot_update_timestamp)
108
+ # save updates
109
+ fd .truncate (0 ) # clear existing content
110
+ fd .seek (0 )
111
+ json .dump (content , fd , indent = 2 )
112
+ fd .write ("\n " ) # add trailing newline for linter and Git compliance
91
113
92
114
93
115
@pytest .hookimpl
0 commit comments