1
1
import dataclasses
2
+ import enum
2
3
import glob
3
4
import mmap
4
5
import typing
@@ -33,6 +34,15 @@ def file_contains_3x_spec_version(document_path: str) -> bool:
33
34
yaml .constructor .SafeConstructor .construct_yaml_str
34
35
)
35
36
37
+ class ResponseType (enum .Enum ):
38
+ DEFAULT_ONLY = enum .auto ()
39
+ STATUS_ONLY = enum .auto ()
40
+ WILDCARD_ONLY = enum .auto ()
41
+ DEFAULT_STATUS = enum .auto ()
42
+ DEFAULT_WILDCARD = enum .auto ()
43
+ STATUS_WIDLCARD = enum .auto ()
44
+ DEFAULT_STATUS_WILDCARD = enum .auto ()
45
+
36
46
@dataclasses .dataclass
37
47
class MetricsData :
38
48
properties_key_qty : int = 0
@@ -43,6 +53,8 @@ class MetricsData:
43
53
required_not_adjacent_to_type_qty : int = 0
44
54
required_adjacent_to_type : typing .Dict [str , int ] = dataclasses .field (default_factory = lambda : {})
45
55
required_key_to_qty : typing .Dict [str , int ] = dataclasses .field (default_factory = lambda : {})
56
+ responses_qty : int = 0
57
+ responses_qtys : typing .Dict [ResponseType , int ] = dataclasses .field (default_factory = lambda : {})
46
58
47
59
48
60
@@ -110,19 +122,61 @@ def __init__(self, stream, metrics_data: MetricsData):
110
122
resolver .Resolver .__init__ (self )
111
123
112
124
113
- def yaml_loading_works (document_path : str , metrics_data : MetricsData ) -> bool :
125
+ def get_yaml_doc (document_path : str , metrics_data : MetricsData ) -> typing . Optional [ dict ] :
114
126
try :
115
127
with open (document_path , 'r' ) as file :
116
128
117
129
loader = CustomLoader (file , metrics_data )
118
130
try :
119
- _loaded_data = loader .get_single_data ()
131
+ return loader .get_single_data ()
120
132
finally :
121
133
loader .dispose ()
122
- return True
123
134
except Exception as exc :
124
135
print (f"Yaml error in file={ document_path } error={ exc } " )
125
- return False
136
+ return None
137
+
138
+ def increment_response_type (metrics_data : MetricsData , response_type : ResponseType ):
139
+ if response_type not in metrics_data .responses_qtys :
140
+ metrics_data .responses_qtys [response_type ] = 0
141
+ metrics_data .responses_qtys [response_type ] += 1
142
+
143
+ def anaylze_doc (yaml_doc : dict , metrics_data : MetricsData ):
144
+ path_items = yaml_doc .get ('paths' )
145
+ if path_items is not None :
146
+ for path_item in path_items .values ():
147
+ if not isinstance (path_item , dict ):
148
+ continue
149
+ for verb in {'get' , 'put' , 'post' , 'delete' , 'options' , 'head' , 'patch' , 'trace' }:
150
+ operation = path_item .get (verb )
151
+ if operation is None :
152
+ continue
153
+ responses : typing .Optional [typing .Dict [str , dict ]] = operation .get ('responses' )
154
+ if responses is None :
155
+ continue
156
+ if len (responses ) == 0 :
157
+ continue
158
+ metrics_data .responses_qty += 1
159
+ if len (responses ) == 1 :
160
+ key = [k for k in responses ][0 ]
161
+ if key == 'default' :
162
+ increment_response_type (metrics_data , ResponseType .DEFAULT_ONLY )
163
+ elif key .endswith ('XX' ):
164
+ increment_response_type (metrics_data , ResponseType .WILDCARD_ONLY )
165
+ else :
166
+ increment_response_type (metrics_data , ResponseType .STATUS_ONLY )
167
+ else :
168
+ default_present = 'default' in responses
169
+ wildcard_present = any (k .endswith ('XX' ) for k in responses )
170
+ status_present = any (not (k .endswith ('XX' ) or k == 'default' ) for k in responses )
171
+ number = [default_present , wildcard_present , status_present ]
172
+ if number == [True , True , False ]:
173
+ increment_response_type (metrics_data , ResponseType .DEFAULT_WILDCARD )
174
+ elif number == [True , False , True ]:
175
+ increment_response_type (metrics_data , ResponseType .DEFAULT_STATUS )
176
+ elif number == [False , True , True ]:
177
+ increment_response_type (metrics_data , ResponseType .STATUS_WIDLCARD )
178
+ elif number == [True , True , True ]:
179
+ increment_response_type (metrics_data , ResponseType .DEFAULT_STATUS_WILDCARD )
126
180
127
181
def filter_and_analyze_documents (document_paths : typing .List [str ], metrics_data : MetricsData ) -> typing .List [str ]:
128
182
filtered_paths : typing .List [str ] = []
@@ -132,6 +186,9 @@ def filter_and_analyze_documents(document_paths: typing.List[str], metrics_data:
132
186
is_v3_spec = file_contains_3x_spec_version (document_path )
133
187
# print(f"path={document_path} v3={is_v3_spec}")
134
188
if is_v3_spec :
135
- if yaml_loading_works (document_path , metrics_data ):
136
- filtered_paths .append (document_path )
189
+ yaml_doc = get_yaml_doc (document_path , metrics_data )
190
+ if yaml_doc is None :
191
+ continue
192
+ anaylze_doc (yaml_doc , metrics_data )
193
+ filtered_paths .append (document_path )
137
194
return filtered_paths
0 commit comments