GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — cli/python/codegen/codegen/generated_test_cases.py at main · databricks/cli Expand file tree/ Copy pathgenerated_test_cases.pyMore file actions 358 lines (272 loc) · 12.5 KB / Copy pathgenerated_test_cases.pyFile metadata and controls358 lines (272 loc) · 12.5 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 """ Generates the per-resource ResourceTestCase data driving databricks_tests/core/test_resources.py. For every wired resource a file _generated/<plural>.py is written (rendered from test_case.py.tmpl) exposing _test_case() -> (ResourceTestCase, _ResourceType). The generated _generated/__init__.py collects them into `test_cases`, which test_resources.py imports and parametrizes its per-resource tests off. dict_example and dataclass_example are synthesized from one value tree and rendered two independent ways -- a dict literal and a constructor expression -- so the dict->dataclass _transform assertion in test_resources.py stays meaningful (the two forms don't share the runtime transform path). Field policy: all required fields (fully expanded), plus optional composite fields (nested dataclass / list / map / enum) on the resource itself; nested objects contribute only their required fields, which keeps examples bounded and avoids recursive schemas (e.g. jobs Task -> ForEachTask -> Task, reachable only through an optional field). Optional scalar, deprecated, and experimental fields are omitted. """ from dataclasses import dataclass from pathlib import Path from string import Template from typing import Union import codegen.jsonschema as openapi import codegen.packages as packages from codegen.generated_enum import _camel_to_upper_snake from codegen.generated_wiring import _wired_resources, _WiredResource HEADER = "# Code generated by pydabs-codegen. DO NOT EDIT.\n\n" _TEST_CASE_TEMPLATE = Template( (Path(__file__).parent / "test_case.py.tmpl").read_text() ) # Launch-stage maturity, mirroring internal/clijson/launchstage.go: # GA < PUBLIC_PREVIEW < PUBLIC_BETA < PRIVATE_PREVIEW; absent stage = GA. _STAGE_RANK = { None: 0, openapi.LaunchStage.GA: 0, openapi.LaunchStage.PUBLIC_PREVIEW: 1, openapi.LaunchStage.PUBLIC_BETA: 2, openapi.LaunchStage.PRIVATE_PREVIEW: 3, } # Synthesized value tree. Each node renders both as a dict literal (dict_example) # and as a constructor expression (dataclass_example). @dataclass class _Scalar: dict_src: str dataclass_src: str @dataclass class _Enum: value: str class_name: str module: str member: str @dataclass class _Object: class_name: str module: str fields: "list[tuple[str, _Value]]" @dataclass class _List: item: "_Value" @dataclass class _Map: key: str value: "_Value" _Value = Union[_Scalar, _Enum, _Object, _List, _Map] def _ref_name(ref: str) -> str: """Last path segment of a JSON-schema ref -- the schema name. :param ref: a JSON-schema reference, e.g. "#/$defs/.../jobs.Task" or "#/$defs/string". """ return ref.split("/")[-1] def _is_composite(ref: str) -> bool: """Whether a ref is a composite type (list, map, object, or enum) rather than a scalar. :param ref: the JSON-schema reference of a field's type. """ if ref.startswith(("#/$defs/slice/", "#/$defs/map/")): return True return _ref_name(ref) not in packages.PRIMITIVES def _synth_scalar(name: str, hint: str) -> _Scalar: """Placeholder value for a primitive (str -> hint, int -> 0, float -> 0.0, bool -> True, time.Time -> RFC3339 string). :param name: the primitive's schema name, e.g. "string", "int", "boolean". :param hint: enclosing field name, used as the string placeholder so examples read meaningfully. """ if name == "string": return _Scalar(f'"{hint}"', f'"{hint}"') if name in ("integer", "int", "int64"): return _Scalar("0", "0") if name in ("number", "float", "float64"): return _Scalar("0.0", "0.0") if name in ("boolean", "bool"): return _Scalar("True", "True") # time.Time is generated as a str (see packages.RENAMES); serialized as RFC3339. if name == "time.Time": return _Scalar('"2020-01-01T00:00:00Z"', '"2020-01-01T00:00:00Z"') # duration.Duration is generated as a str; serialized as a seconds string. if name == "duration.Duration": return _Scalar('"3600s"', '"3600s"') # interface{} is generated as Any; there is no schema to sample, so use a # fixed inline dict. Both render paths use the same literal. if name == "interface": return _Scalar('{"key": "value"}', '{"key": "value"}') raise ValueError(f"Unknown primitive: {name}") def _synth_ref( namespace: str, ref: str, hint: str, schemas: dict[str, openapi.Schema], visiting: set[str], ) -> _Value: """Synthesize a value node for whatever type a ref points at: list, map, scalar, enum, or nested object. :param namespace: the resource's namespace (e.g. "jobs"); selects the module a referenced type is generated into. :param ref: the JSON-schema reference of the type to synthesize. :param hint: enclosing field name, passed through as the string placeholder. :param schemas: all post-patch schemas keyed by schema name, for looking up nested/enum types. :param visiting: ancestor object names on the current path, used to detect required cycles. """ if ref.startswith("#/$defs/slice/"): element_ref = ref.replace("#/$defs/slice/", "#/$defs/") return _List(_synth_ref(namespace, element_ref, hint, schemas, visiting)) if ref.startswith("#/$defs/map/"): # generate_type only ever produces dict[str, str] maps (map/string). if ref != "#/$defs/map/string": raise ValueError(f"Unsupported map ref: {ref}") return _Map("key", _Scalar('"value"', '"value"')) name = _ref_name(ref) if name in packages.PRIMITIVES: return _synth_scalar(name, hint) schema = schemas[name] class_name = packages.get_class_name(ref) module = _module_of(namespace, ref) if schema.type == openapi.SchemaType.STRING: value = schema.enum[0] return _Enum(value, class_name, module, _camel_to_upper_snake(value)) # Only reachable through required fields at this depth (see _synth_object); a # required cycle has no finite value, so fail loudly instead of looping. if name in visiting: raise ValueError(f"Required-field cycle through '{name}'") return _synth_object(namespace, name, schema, schemas, visiting, top_level=False) def _synth_object( namespace: str, schema_name: str, schema: openapi.Schema, schemas: dict[str, openapi.Schema], visiting: set[str], top_level: bool, ) -> _Object: """Synthesize an object value, choosing fields by policy: all required fields, plus (only at the resource top level) stable optional composite fields. :param namespace: the resource's namespace, threaded through to resolve nested types' modules. :param schema_name: this object's schema name (e.g. "resources.Alert"). :param schema: the Schema for this object -- its properties and required list. :param schemas: all post-patch schemas, for recursing into nested types. :param visiting: ancestor object names on the current path (cycle guard). :param top_level: True only for the resource itself; when False, all optional fields are dropped. """ visiting = visiting | {schema_name} fields: list[tuple[str, _Value]] = [] for field_name, prop in schema.properties.items(): required = field_name in schema.required if not required: # Nested objects contribute only required fields; on the resource # itself, also include stable optional composite fields. if not top_level: continue if not _is_composite(prop.ref): continue if ( prop.deprecated or _STAGE_RANK.get(prop.stage, 0) > _STAGE_RANK[openapi.LaunchStage.PUBLIC_PREVIEW] ): continue value = _synth_ref(namespace, prop.ref, field_name, schemas, visiting) fields.append((field_name, value)) return _Object( packages.get_class_name(schema_name), _module_of(namespace, schema_name), fields ) def _module_of(namespace: str, schema_name: str) -> str: """Python module a (non-primitive) schema's generated class lives in; asserts it exists. :param namespace: the resource's namespace; the type is generated under databricks.bundles.<namespace>._models. :param schema_name: the object/enum schema name to resolve. """ module = packages.get_package(namespace, schema_name) assert module return module def _render_dict(value: _Value) -> str: """Render a synthesized value as a dict-literal source string (the dict_example form). :param value: the synthesized value node to render. """ if isinstance(value, _Scalar): return value.dict_src if isinstance(value, _Enum): return f'"{value.value}"' if isinstance(value, _List): return f"[{_render_dict(value.item)}]" if isinstance(value, _Map): return "{" + f'"{value.key}": {_render_dict(value.value)}' + "}" fields = ", ".join( f'"{name}": {_render_dict(child)}' for name, child in value.fields ) return "{" + fields + "}" def _render_dataclass(value: _Value) -> str: """Render a synthesized value as a constructor-expression source string (the dataclass_example form). :param value: the synthesized value node to render. """ if isinstance(value, _Scalar): return value.dataclass_src if isinstance(value, _Enum): return f"{value.class_name}.{value.member}" if isinstance(value, _List): return f"[{_render_dataclass(value.item)}]" if isinstance(value, _Map): return "{" + f'"{value.key}": {_render_dataclass(value.value)}' + "}" fields = ", ".join( f"{name}={_render_dataclass(child)}" for name, child in value.fields ) return f"{value.class_name}({fields})" def _collect_imports(value: _Value, out: set[tuple[str, str]]) -> None: """Collect (module, class_name) pairs the dataclass_example needs, walking nested objects/enums. :param value: the synthesized value node to walk. :param out: set accumulating the (module, class_name) import pairs; mutated in place. """ if isinstance(value, _Enum): out.add((value.module, value.class_name)) elif isinstance(value, _Object): out.add((value.module, value.class_name)) for _, child in value.fields: _collect_imports(child, out) elif isinstance(value, _List): _collect_imports(value.item, out) elif isinstance(value, _Map): _collect_imports(value.value, out) def write_test_cases(output: str, schemas: dict[str, openapi.Schema]): """Write one _generated/<resource_plural>.py per wired resource plus the collector __init__.py. :param output: codegen output root (the python/ directory); files land under databricks_tests/core/_generated. :param schemas: all post-patch schemas, used to synthesize each resource's dict/dataclass examples. """ resources = _wired_resources() generated_path = Path(output) / "databricks_tests" / "core" / "_generated" generated_path.mkdir(parents=True, exist_ok=True) plural_to_ref = {ns: ref for ref, ns in packages.RESOURCE_NAMESPACE.items()} for r in resources: resource_ref = plural_to_ref[r.plural_name] schema = schemas[resource_ref] example = _synth_object( r.plural_name, resource_ref, schema, schemas, set(), top_level=True ) imports: set[tuple[str, str]] = set() _collect_imports(example, imports) model_imports = "\n".join( f"from {module} import {class_name}" for module, class_name in sorted(imports) ) code = _TEST_CASE_TEMPLATE.substitute( singular=r.singular_name, plural=r.plural_name, model_imports=model_imports, dict_example=_render_dict(example), dataclass_example=_render_dataclass(example), ) (generated_path / f"{r.plural_name}.py").write_text(HEADER + code) (generated_path / "__init__.py").write_text(HEADER + _collector_code(resources)) print(f"Writing test cases into {generated_path}") def _collector_code(resources: list[_WiredResource]) -> str: """Source for _generated/__init__.py: imports the per-resource modules and assembles `test_cases`. :param resources: the wired resources, in the order their test cases are collected. """ module_imports = "\n".join(f" {r.plural_name}," for r in resources) entries = "\n".join(f" {r.plural_name}._test_case()," for r in resources) return f"""from databricks_tests.core._generated import ( {module_imports} ) __all__ = ["test_cases"] test_cases = [ {entries} ] """ |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |