(self)
| 155 | return scenario_def["tests"] |
| 156 | |
| 157 | def _create_tests(self): |
| 158 | for dirpath, _, filenames in os.walk(self.test_path): |
| 159 | dirname = os.path.split(dirpath)[-1] |
| 160 | |
| 161 | for filename in filenames: |
| 162 | with open(os.path.join(dirpath, filename)) as scenario_stream: # noqa: ASYNC101, RUF100 |
| 163 | # Use tz_aware=False to match how CodecOptions decodes |
| 164 | # dates. |
| 165 | opts = json_util.JSONOptions(tz_aware=False) |
| 166 | scenario_def = ScenarioDict( |
| 167 | json_util.loads(scenario_stream.read(), json_options=opts) |
| 168 | ) |
| 169 | |
| 170 | test_type = os.path.splitext(filename)[0] |
| 171 | |
| 172 | # Construct test from scenario. |
| 173 | for test_def in self.tests(scenario_def): |
| 174 | test_name = "test_{}_{}_{}".format( |
| 175 | dirname, |
| 176 | test_type.replace("-", "_").replace(".", "_"), |
| 177 | str(test_def["description"].replace(" ", "_").replace(".", "_")), |
| 178 | ) |
| 179 | |
| 180 | new_test = self._create_test(scenario_def, test_def, test_name) |
| 181 | new_test = self._ensure_min_max_server_version(scenario_def, new_test) |
| 182 | new_test = self.ensure_run_on(scenario_def, new_test) |
| 183 | |
| 184 | new_test.__name__ = test_name |
| 185 | setattr(self._test_class, new_test.__name__, new_test) |
| 186 | |
| 187 | def create_tests(self): |
| 188 | if _IS_SYNC: |
no test coverage detected