We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Set pattern field for `mac` fields
Implement a regex pattern to validate values using the `pattern` attribute from #1834 

Proposed regex: `^[A-F0-9]{2}(-[A-F0-9]{2}){5,}$`

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `scripts/schema/cleaner.py`
Content:
```
1 # Licensed to Elasticsearch B.V. under one or more contributor
2 # license agreements. See the NOTICE file distributed with
3 # this work for additional information regarding copyright
4 # ownership. Elasticsearch B.V. licenses this file to you under
5 # the Apache License, Version 2.0 (the "License"); you may
6 # not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # 	http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 
18 import re
19 
20 from generators import ecs_helpers
21 from schema import visitor
22 
23 # This script performs a few cleanup functions in place, within the deeply nested
24 # 'fields' structure passed to `clean(fields)`.
25 #
26 # What happens here:
27 #
28 # - check that mandatory attributes are present, without which we can't do much.
29 # - cleans things up, like stripping spaces, sorting arrays
30 # - makes lots of defaults explicit
31 # - pre-calculate a few additional helpful fields
32 # - converts shorthands into full representation (e.g. reuse locations)
33 #
34 # This script only deals with field sets themselves and the fields defined
35 # inside them. It doesn't perform field reuse, and therefore doesn't
36 # deal with final field names either.
37 
38 
39 def clean(fields, strict=False):
40     global strict_mode
41     strict_mode = strict
42     visitor.visit_fields(fields, fieldset_func=schema_cleanup, field_func=field_cleanup)
43 
44 
45 # Schema level cleanup
46 
47 
48 def schema_cleanup(schema):
49     # Sanity check first
50     schema_mandatory_attributes(schema)
51     # trailing space cleanup
52     ecs_helpers.dict_clean_string_values(schema['schema_details'])
53     ecs_helpers.dict_clean_string_values(schema['field_details'])
54     # Some defaults
55     schema['schema_details'].setdefault('group', 2)
56     schema['schema_details'].setdefault('root', False)
57     schema['field_details'].setdefault('type', 'group')
58     schema['field_details'].setdefault('short', schema['field_details']['description'])
59     if 'reusable' in schema['schema_details']:
60         # order to perform chained reuses. Set to 1 if it needs to happen earlier.
61         schema['schema_details']['reusable'].setdefault('order', 2)
62     # Precalculate stuff. Those can't be set in the YAML.
63     if schema['schema_details']['root']:
64         schema['schema_details']['prefix'] = ''
65     else:
66         schema['schema_details']['prefix'] = schema['field_details']['name'] + '.'
67     normalize_reuse_notation(schema)
68     # Final validity check if in strict mode
69     schema_assertions_and_warnings(schema)
70 
71 
72 SCHEMA_MANDATORY_ATTRIBUTES = ['name', 'title', 'description']
73 
74 
75 def schema_mandatory_attributes(schema):
76     """Ensures for the presence of the mandatory schema attributes and raises if any are missing"""
77     current_schema_attributes = sorted(list(schema['field_details'].keys()) +
78                                        list(schema['schema_details'].keys()))
79     missing_attributes = ecs_helpers.list_subtract(SCHEMA_MANDATORY_ATTRIBUTES, current_schema_attributes)
80     if len(missing_attributes) > 0:
81         msg = "Schema {} is missing the following mandatory attributes: {}.\nFound these: {}".format(
82             schema['field_details']['name'], ', '.join(missing_attributes), current_schema_attributes)
83         raise ValueError(msg)
84     if 'reusable' in schema['schema_details']:
85         reuse_attributes = sorted(schema['schema_details']['reusable'].keys())
86         missing_reuse_attributes = ecs_helpers.list_subtract(['expected', 'top_level'], reuse_attributes)
87         if len(missing_reuse_attributes) > 0:
88             msg = "Reusable schema {} is missing the following reuse attributes: {}.\nFound these: {}".format(
89                 schema['field_details']['name'], ', '.join(missing_reuse_attributes), reuse_attributes)
90             raise ValueError(msg)
91 
92 
93 def schema_assertions_and_warnings(schema):
94     """Additional checks on a fleshed out schema"""
95     single_line_short_description(schema, strict=strict_mode)
96     if 'beta' in schema['field_details']:
97         single_line_beta_description(schema, strict=strict_mode)
98     if 'reusable' in schema['schema_details']:
99         single_line_short_override_description(schema, strict=strict_mode)
100 
101 
102 def normalize_reuse_notation(schema):
103     """
104     Replace single word reuse shorthands from the schema YAMLs with the explicit {at: , as:} notation.
105 
106     When marking "user" as reusable under "destination" with the shorthand entry
107     `- destination`, this is expanded to the complete entry
108     `- { "at": "destination", "as": "user" }`.
109     The field set is thus nested at `destination.user.*`, with fields such as `destination.user.name`.
110 
111     The dictionary notation enables nesting a field set as a different name.
112     An example is nesting "process" fields to capture parent process details
113     at `process.parent.*`.
114     The dictionary notation `- { "at": "process", "as": "parent" }` will yield
115     fields such as `process.parent.pid`.
116     """
117     if 'reusable' not in schema['schema_details']:
118         return
119     schema_name = schema['field_details']['name']
120     reuse_entries = []
121     for reuse_entry in schema['schema_details']['reusable']['expected']:
122         if type(reuse_entry) is dict:  # Already explicit
123             if 'at' in reuse_entry and 'as' in reuse_entry:
124                 explicit_entry = reuse_entry
125             else:
126                 raise ValueError("When specifying reusable expected locations for {} " +
127                                  "with the dictionary notation, keys 'as' and 'at' are required. " +
128                                  "Got {}.".format(schema_name, reuse_entry))
129         else:  # Make it explicit
130             explicit_entry = {'at': reuse_entry, 'as': schema_name}
131         explicit_entry['full'] = explicit_entry['at'] + '.' + explicit_entry['as']
132         reuse_entries.append(explicit_entry)
133     schema['schema_details']['reusable']['expected'] = reuse_entries
134 
135 
136 # Field level cleanup
137 
138 
139 def field_cleanup(field):
140     field_mandatory_attributes(field)
141     if ecs_helpers.is_intermediate(field):
142         return
143     ecs_helpers.dict_clean_string_values(field['field_details'])
144     if 'allowed_values' in field['field_details']:
145         for allowed_value in field['field_details']['allowed_values']:
146             ecs_helpers.dict_clean_string_values(allowed_value)
147     field_defaults(field)
148     field_assertions_and_warnings(field)
149 
150 
151 def field_defaults(field):
152     field['field_details'].setdefault('short', field['field_details']['description'])
153     field['field_details'].setdefault('normalize', [])
154     field_or_multi_field_datatype_defaults(field['field_details'])
155     if 'multi_fields' in field['field_details']:
156         for mf in field['field_details']['multi_fields']:
157             field_or_multi_field_datatype_defaults(mf)
158             if 'name' not in mf:
159                 mf['name'] = mf['type']
160 
161 
162 def field_or_multi_field_datatype_defaults(field_details):
163     """Sets datatype-related defaults on a canonical field or multi-field entries."""
164     if field_details['type'] == 'keyword':
165         field_details.setdefault('ignore_above', 1024)
166     if field_details['type'] == 'text':
167         field_details.setdefault('norms', False)
168     # wildcard needs the index param stripped
169     if field_details['type'] == 'wildcard':
170         field_details.pop('index', None)
171     if 'index' in field_details and not field_details['index']:
172         field_details.setdefault('doc_values', False)
173         field_details.pop('ignore_above', None)
174 
175 
176 FIELD_MANDATORY_ATTRIBUTES = ['name', 'description', 'type', 'level']
177 ACCEPTABLE_FIELD_LEVELS = ['core', 'extended', 'custom']
178 
179 
180 def field_mandatory_attributes(field):
181     """Ensures for the presence of the mandatory field attributes and raises if any are missing"""
182     if ecs_helpers.is_intermediate(field):
183         return
184     current_field_attributes = sorted(field['field_details'].keys())
185     missing_attributes = ecs_helpers.list_subtract(FIELD_MANDATORY_ATTRIBUTES, current_field_attributes)
186 
187     # `alias` fields require a target `path` attribute.
188     if field['field_details'].get('type') == 'alias' and 'path' not in current_field_attributes:
189         missing_attributes.append('path')
190     # `scaled_float` fields require a `scaling_factor` attribute.
191     if field['field_details'].get('type') == 'scaled_float' and 'scaling_factor' not in current_field_attributes:
192         missing_attributes.append('scaling_factor')
193 
194     if len(missing_attributes) > 0:
195         msg = "Field is missing the following mandatory attributes: {}.\nFound these: {}.\nField details: {}"
196         raise ValueError(msg.format(', '.join(missing_attributes),
197                                     current_field_attributes, field))
198 
199 
200 def field_assertions_and_warnings(field):
201     """Additional checks on a fleshed out field"""
202     if not ecs_helpers.is_intermediate(field):
203         # check short description length if in strict mode
204         single_line_short_description(field, strict=strict_mode)
205         if 'beta' in field['field_details']:
206             single_line_beta_description(field, strict=strict_mode)
207         if 'pattern' in field['field_details']:
208             validate_pattern_regex(field['field_details'], strict=strict_mode)
209         check_example_value(field, strict=strict_mode)
210         if field['field_details']['level'] not in ACCEPTABLE_FIELD_LEVELS:
211             msg = "Invalid level for field '{}'.\nValue: {}\nAcceptable values: {}".format(
212                 field['field_details']['name'], field['field_details']['level'],
213                 ACCEPTABLE_FIELD_LEVELS)
214             raise ValueError(msg)
215 
216 # Common
217 
218 
219 SHORT_LIMIT = 120
220 
221 
222 def single_line_short_check(short_to_check, short_name):
223     short_length = len(short_to_check)
224     if "\n" in short_to_check or short_length > SHORT_LIMIT:
225         msg = "Short descriptions must be single line, and under {} characters (current length: {}).\n".format(
226             SHORT_LIMIT, short_length)
227         msg += "Offending field or field set: {}\nShort description:\n  {}".format(
228             short_name,
229             short_to_check)
230         return msg
231     return None
232 
233 
234 def strict_warning_handler(message, strict):
235     """Handles warnings based on --strict mode"""
236     if strict:
237         raise ValueError(message)
238     else:
239         ecs_helpers.strict_warning(message)
240 
241 
242 def single_line_short_description(schema_or_field, strict=True):
243     error = single_line_short_check(schema_or_field['field_details']['short'], schema_or_field['field_details']['name'])
244     if error:
245         strict_warning_handler(error, strict)
246 
247 
248 def single_line_short_override_description(schema_or_field, strict=True):
249     for field in schema_or_field['schema_details']['reusable']['expected']:
250         if not 'short_override' in field:
251             continue
252         error = single_line_short_check(field['short_override'], field['full'])
253         if error:
254             strict_warning_handler(error, strict)
255 
256 
257 def check_example_value(field, strict=True):
258     """
259     Checks if value of the example field is of type list or dict.
260     Fails or warns (depending on strict mode) if so.
261     """
262     example_value = field['field_details'].get('example', None)
263     pattern = field['field_details'].get('pattern', None)
264     name = field['field_details']['name']
265 
266     if isinstance(example_value, (list, dict)):
267         msg = f"Example value for field `{name}` contains an object or array which must be quoted to avoid YAML interpretation."
268         strict_warning_handler(msg, strict)
269 
270     if pattern:
271         match = re.match(pattern, example_value)
272         if not match:
273             msg = f"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`."
274             strict_warning_handler(msg, strict)
275 
276 
277 def single_line_beta_description(schema_or_field, strict=True):
278     if "\n" in schema_or_field['field_details']['beta']:
279         msg = "Beta descriptions must be single line.\n"
280         msg += f"Offending field or field set: {schema_or_field['field_details']['name']}"
281         strict_warning_handler(msg, strict)
282 
283 
284 def validate_pattern_regex(field, strict=True):
285     """
286     Validates if field['pattern'] contains a valid regular expression.
287     """
288     try:
289         re.compile(field['pattern'])
290     except re.error:
291         msg = "Pattern value must be a valid regular expression.\n"
292         msg += f"Offending field name: {field['name']}"
293         strict_warning_handler(msg, strict)
294
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/scripts/schema/cleaner.py b/scripts/schema/cleaner.py
--- a/scripts/schema/cleaner.py
+++ b/scripts/schema/cleaner.py
@@ -268,10 +268,18 @@
         strict_warning_handler(msg, strict)
 
     if pattern:
-        match = re.match(pattern, example_value)
-        if not match:
-            msg = f"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`."
-            strict_warning_handler(msg, strict)
+        # Examples with arrays must be handled
+        if 'array' in field['field_details']['normalize']:
+            # strips unnecessary chars in order to split each example value
+            example_values = example_value.translate(str.maketrans('', '', '"[] ')).split(',')
+        else:
+            example_values = [example_value]
+
+        for example_value in example_values:
+            match = re.match(pattern, example_value)
+            if not match:
+                msg = f"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`."
+                strict_warning_handler(msg, strict)
 
 
 def single_line_beta_description(schema_or_field, strict=True):

{"golden_diff": "diff --git a/scripts/schema/cleaner.py b/scripts/schema/cleaner.py\n--- a/scripts/schema/cleaner.py\n+++ b/scripts/schema/cleaner.py\n@@ -268,10 +268,18 @@\n         strict_warning_handler(msg, strict)\n \n     if pattern:\n-        match = re.match(pattern, example_value)\n-        if not match:\n-            msg = f\"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`.\"\n-            strict_warning_handler(msg, strict)\n+        # Examples with arrays must be handled\n+        if 'array' in field['field_details']['normalize']:\n+            # strips unnecessary chars in order to split each example value\n+            example_values = example_value.translate(str.maketrans('', '', '\"[] ')).split(',')\n+        else:\n+            example_values = [example_value]\n+\n+        for example_value in example_values:\n+            match = re.match(pattern, example_value)\n+            if not match:\n+                msg = f\"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`.\"\n+                strict_warning_handler(msg, strict)\n \n \n def single_line_beta_description(schema_or_field, strict=True):\n", "issue": "Set pattern field for `mac` fields\nImplement a regex pattern to validate values using the `pattern` attribute from #1834 \r\n\r\nProposed regex: `^[A-F0-9]{2}(-[A-F0-9]{2}){5,}$`\n", "before_files": [{"content": "# Licensed to Elasticsearch B.V. under one or more contributor\n# license agreements. See the NOTICE file distributed with\n# this work for additional information regarding copyright\n# ownership. Elasticsearch B.V. licenses this file to you under\n# the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# \thttp://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport re\n\nfrom generators import ecs_helpers\nfrom schema import visitor\n\n# This script performs a few cleanup functions in place, within the deeply nested\n# 'fields' structure passed to `clean(fields)`.\n#\n# What happens here:\n#\n# - check that mandatory attributes are present, without which we can't do much.\n# - cleans things up, like stripping spaces, sorting arrays\n# - makes lots of defaults explicit\n# - pre-calculate a few additional helpful fields\n# - converts shorthands into full representation (e.g. reuse locations)\n#\n# This script only deals with field sets themselves and the fields defined\n# inside them. It doesn't perform field reuse, and therefore doesn't\n# deal with final field names either.\n\n\ndef clean(fields, strict=False):\n    global strict_mode\n    strict_mode = strict\n    visitor.visit_fields(fields, fieldset_func=schema_cleanup, field_func=field_cleanup)\n\n\n# Schema level cleanup\n\n\ndef schema_cleanup(schema):\n    # Sanity check first\n    schema_mandatory_attributes(schema)\n    # trailing space cleanup\n    ecs_helpers.dict_clean_string_values(schema['schema_details'])\n    ecs_helpers.dict_clean_string_values(schema['field_details'])\n    # Some defaults\n    schema['schema_details'].setdefault('group', 2)\n    schema['schema_details'].setdefault('root', False)\n    schema['field_details'].setdefault('type', 'group')\n    schema['field_details'].setdefault('short', schema['field_details']['description'])\n    if 'reusable' in schema['schema_details']:\n        # order to perform chained reuses. Set to 1 if it needs to happen earlier.\n        schema['schema_details']['reusable'].setdefault('order', 2)\n    # Precalculate stuff. Those can't be set in the YAML.\n    if schema['schema_details']['root']:\n        schema['schema_details']['prefix'] = ''\n    else:\n        schema['schema_details']['prefix'] = schema['field_details']['name'] + '.'\n    normalize_reuse_notation(schema)\n    # Final validity check if in strict mode\n    schema_assertions_and_warnings(schema)\n\n\nSCHEMA_MANDATORY_ATTRIBUTES = ['name', 'title', 'description']\n\n\ndef schema_mandatory_attributes(schema):\n    \"\"\"Ensures for the presence of the mandatory schema attributes and raises if any are missing\"\"\"\n    current_schema_attributes = sorted(list(schema['field_details'].keys()) +\n                                       list(schema['schema_details'].keys()))\n    missing_attributes = ecs_helpers.list_subtract(SCHEMA_MANDATORY_ATTRIBUTES, current_schema_attributes)\n    if len(missing_attributes) > 0:\n        msg = \"Schema {} is missing the following mandatory attributes: {}.\\nFound these: {}\".format(\n            schema['field_details']['name'], ', '.join(missing_attributes), current_schema_attributes)\n        raise ValueError(msg)\n    if 'reusable' in schema['schema_details']:\n        reuse_attributes = sorted(schema['schema_details']['reusable'].keys())\n        missing_reuse_attributes = ecs_helpers.list_subtract(['expected', 'top_level'], reuse_attributes)\n        if len(missing_reuse_attributes) > 0:\n            msg = \"Reusable schema {} is missing the following reuse attributes: {}.\\nFound these: {}\".format(\n                schema['field_details']['name'], ', '.join(missing_reuse_attributes), reuse_attributes)\n            raise ValueError(msg)\n\n\ndef schema_assertions_and_warnings(schema):\n    \"\"\"Additional checks on a fleshed out schema\"\"\"\n    single_line_short_description(schema, strict=strict_mode)\n    if 'beta' in schema['field_details']:\n        single_line_beta_description(schema, strict=strict_mode)\n    if 'reusable' in schema['schema_details']:\n        single_line_short_override_description(schema, strict=strict_mode)\n\n\ndef normalize_reuse_notation(schema):\n    \"\"\"\n    Replace single word reuse shorthands from the schema YAMLs with the explicit {at: , as:} notation.\n\n    When marking \"user\" as reusable under \"destination\" with the shorthand entry\n    `- destination`, this is expanded to the complete entry\n    `- { \"at\": \"destination\", \"as\": \"user\" }`.\n    The field set is thus nested at `destination.user.*`, with fields such as `destination.user.name`.\n\n    The dictionary notation enables nesting a field set as a different name.\n    An example is nesting \"process\" fields to capture parent process details\n    at `process.parent.*`.\n    The dictionary notation `- { \"at\": \"process\", \"as\": \"parent\" }` will yield\n    fields such as `process.parent.pid`.\n    \"\"\"\n    if 'reusable' not in schema['schema_details']:\n        return\n    schema_name = schema['field_details']['name']\n    reuse_entries = []\n    for reuse_entry in schema['schema_details']['reusable']['expected']:\n        if type(reuse_entry) is dict:  # Already explicit\n            if 'at' in reuse_entry and 'as' in reuse_entry:\n                explicit_entry = reuse_entry\n            else:\n                raise ValueError(\"When specifying reusable expected locations for {} \" +\n                                 \"with the dictionary notation, keys 'as' and 'at' are required. \" +\n                                 \"Got {}.\".format(schema_name, reuse_entry))\n        else:  # Make it explicit\n            explicit_entry = {'at': reuse_entry, 'as': schema_name}\n        explicit_entry['full'] = explicit_entry['at'] + '.' + explicit_entry['as']\n        reuse_entries.append(explicit_entry)\n    schema['schema_details']['reusable']['expected'] = reuse_entries\n\n\n# Field level cleanup\n\n\ndef field_cleanup(field):\n    field_mandatory_attributes(field)\n    if ecs_helpers.is_intermediate(field):\n        return\n    ecs_helpers.dict_clean_string_values(field['field_details'])\n    if 'allowed_values' in field['field_details']:\n        for allowed_value in field['field_details']['allowed_values']:\n            ecs_helpers.dict_clean_string_values(allowed_value)\n    field_defaults(field)\n    field_assertions_and_warnings(field)\n\n\ndef field_defaults(field):\n    field['field_details'].setdefault('short', field['field_details']['description'])\n    field['field_details'].setdefault('normalize', [])\n    field_or_multi_field_datatype_defaults(field['field_details'])\n    if 'multi_fields' in field['field_details']:\n        for mf in field['field_details']['multi_fields']:\n            field_or_multi_field_datatype_defaults(mf)\n            if 'name' not in mf:\n                mf['name'] = mf['type']\n\n\ndef field_or_multi_field_datatype_defaults(field_details):\n    \"\"\"Sets datatype-related defaults on a canonical field or multi-field entries.\"\"\"\n    if field_details['type'] == 'keyword':\n        field_details.setdefault('ignore_above', 1024)\n    if field_details['type'] == 'text':\n        field_details.setdefault('norms', False)\n    # wildcard needs the index param stripped\n    if field_details['type'] == 'wildcard':\n        field_details.pop('index', None)\n    if 'index' in field_details and not field_details['index']:\n        field_details.setdefault('doc_values', False)\n        field_details.pop('ignore_above', None)\n\n\nFIELD_MANDATORY_ATTRIBUTES = ['name', 'description', 'type', 'level']\nACCEPTABLE_FIELD_LEVELS = ['core', 'extended', 'custom']\n\n\ndef field_mandatory_attributes(field):\n    \"\"\"Ensures for the presence of the mandatory field attributes and raises if any are missing\"\"\"\n    if ecs_helpers.is_intermediate(field):\n        return\n    current_field_attributes = sorted(field['field_details'].keys())\n    missing_attributes = ecs_helpers.list_subtract(FIELD_MANDATORY_ATTRIBUTES, current_field_attributes)\n\n    # `alias` fields require a target `path` attribute.\n    if field['field_details'].get('type') == 'alias' and 'path' not in current_field_attributes:\n        missing_attributes.append('path')\n    # `scaled_float` fields require a `scaling_factor` attribute.\n    if field['field_details'].get('type') == 'scaled_float' and 'scaling_factor' not in current_field_attributes:\n        missing_attributes.append('scaling_factor')\n\n    if len(missing_attributes) > 0:\n        msg = \"Field is missing the following mandatory attributes: {}.\\nFound these: {}.\\nField details: {}\"\n        raise ValueError(msg.format(', '.join(missing_attributes),\n                                    current_field_attributes, field))\n\n\ndef field_assertions_and_warnings(field):\n    \"\"\"Additional checks on a fleshed out field\"\"\"\n    if not ecs_helpers.is_intermediate(field):\n        # check short description length if in strict mode\n        single_line_short_description(field, strict=strict_mode)\n        if 'beta' in field['field_details']:\n            single_line_beta_description(field, strict=strict_mode)\n        if 'pattern' in field['field_details']:\n            validate_pattern_regex(field['field_details'], strict=strict_mode)\n        check_example_value(field, strict=strict_mode)\n        if field['field_details']['level'] not in ACCEPTABLE_FIELD_LEVELS:\n            msg = \"Invalid level for field '{}'.\\nValue: {}\\nAcceptable values: {}\".format(\n                field['field_details']['name'], field['field_details']['level'],\n                ACCEPTABLE_FIELD_LEVELS)\n            raise ValueError(msg)\n\n# Common\n\n\nSHORT_LIMIT = 120\n\n\ndef single_line_short_check(short_to_check, short_name):\n    short_length = len(short_to_check)\n    if \"\\n\" in short_to_check or short_length > SHORT_LIMIT:\n        msg = \"Short descriptions must be single line, and under {} characters (current length: {}).\\n\".format(\n            SHORT_LIMIT, short_length)\n        msg += \"Offending field or field set: {}\\nShort description:\\n  {}\".format(\n            short_name,\n            short_to_check)\n        return msg\n    return None\n\n\ndef strict_warning_handler(message, strict):\n    \"\"\"Handles warnings based on --strict mode\"\"\"\n    if strict:\n        raise ValueError(message)\n    else:\n        ecs_helpers.strict_warning(message)\n\n\ndef single_line_short_description(schema_or_field, strict=True):\n    error = single_line_short_check(schema_or_field['field_details']['short'], schema_or_field['field_details']['name'])\n    if error:\n        strict_warning_handler(error, strict)\n\n\ndef single_line_short_override_description(schema_or_field, strict=True):\n    for field in schema_or_field['schema_details']['reusable']['expected']:\n        if not 'short_override' in field:\n            continue\n        error = single_line_short_check(field['short_override'], field['full'])\n        if error:\n            strict_warning_handler(error, strict)\n\n\ndef check_example_value(field, strict=True):\n    \"\"\"\n    Checks if value of the example field is of type list or dict.\n    Fails or warns (depending on strict mode) if so.\n    \"\"\"\n    example_value = field['field_details'].get('example', None)\n    pattern = field['field_details'].get('pattern', None)\n    name = field['field_details']['name']\n\n    if isinstance(example_value, (list, dict)):\n        msg = f\"Example value for field `{name}` contains an object or array which must be quoted to avoid YAML interpretation.\"\n        strict_warning_handler(msg, strict)\n\n    if pattern:\n        match = re.match(pattern, example_value)\n        if not match:\n            msg = f\"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`.\"\n            strict_warning_handler(msg, strict)\n\n\ndef single_line_beta_description(schema_or_field, strict=True):\n    if \"\\n\" in schema_or_field['field_details']['beta']:\n        msg = \"Beta descriptions must be single line.\\n\"\n        msg += f\"Offending field or field set: {schema_or_field['field_details']['name']}\"\n        strict_warning_handler(msg, strict)\n\n\ndef validate_pattern_regex(field, strict=True):\n    \"\"\"\n    Validates if field['pattern'] contains a valid regular expression.\n    \"\"\"\n    try:\n        re.compile(field['pattern'])\n    except re.error:\n        msg = \"Pattern value must be a valid regular expression.\\n\"\n        msg += f\"Offending field name: {field['name']}\"\n        strict_warning_handler(msg, strict)\n", "path": "scripts/schema/cleaner.py"}], "after_files": [{"content": "# Licensed to Elasticsearch B.V. under one or more contributor\n# license agreements. See the NOTICE file distributed with\n# this work for additional information regarding copyright\n# ownership. Elasticsearch B.V. licenses this file to you under\n# the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# \thttp://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport re\n\nfrom generators import ecs_helpers\nfrom schema import visitor\n\n# This script performs a few cleanup functions in place, within the deeply nested\n# 'fields' structure passed to `clean(fields)`.\n#\n# What happens here:\n#\n# - check that mandatory attributes are present, without which we can't do much.\n# - cleans things up, like stripping spaces, sorting arrays\n# - makes lots of defaults explicit\n# - pre-calculate a few additional helpful fields\n# - converts shorthands into full representation (e.g. reuse locations)\n#\n# This script only deals with field sets themselves and the fields defined\n# inside them. It doesn't perform field reuse, and therefore doesn't\n# deal with final field names either.\n\n\ndef clean(fields, strict=False):\n    global strict_mode\n    strict_mode = strict\n    visitor.visit_fields(fields, fieldset_func=schema_cleanup, field_func=field_cleanup)\n\n\n# Schema level cleanup\n\n\ndef schema_cleanup(schema):\n    # Sanity check first\n    schema_mandatory_attributes(schema)\n    # trailing space cleanup\n    ecs_helpers.dict_clean_string_values(schema['schema_details'])\n    ecs_helpers.dict_clean_string_values(schema['field_details'])\n    # Some defaults\n    schema['schema_details'].setdefault('group', 2)\n    schema['schema_details'].setdefault('root', False)\n    schema['field_details'].setdefault('type', 'group')\n    schema['field_details'].setdefault('short', schema['field_details']['description'])\n    if 'reusable' in schema['schema_details']:\n        # order to perform chained reuses. Set to 1 if it needs to happen earlier.\n        schema['schema_details']['reusable'].setdefault('order', 2)\n    # Precalculate stuff. Those can't be set in the YAML.\n    if schema['schema_details']['root']:\n        schema['schema_details']['prefix'] = ''\n    else:\n        schema['schema_details']['prefix'] = schema['field_details']['name'] + '.'\n    normalize_reuse_notation(schema)\n    # Final validity check if in strict mode\n    schema_assertions_and_warnings(schema)\n\n\nSCHEMA_MANDATORY_ATTRIBUTES = ['name', 'title', 'description']\n\n\ndef schema_mandatory_attributes(schema):\n    \"\"\"Ensures for the presence of the mandatory schema attributes and raises if any are missing\"\"\"\n    current_schema_attributes = sorted(list(schema['field_details'].keys()) +\n                                       list(schema['schema_details'].keys()))\n    missing_attributes = ecs_helpers.list_subtract(SCHEMA_MANDATORY_ATTRIBUTES, current_schema_attributes)\n    if len(missing_attributes) > 0:\n        msg = \"Schema {} is missing the following mandatory attributes: {}.\\nFound these: {}\".format(\n            schema['field_details']['name'], ', '.join(missing_attributes), current_schema_attributes)\n        raise ValueError(msg)\n    if 'reusable' in schema['schema_details']:\n        reuse_attributes = sorted(schema['schema_details']['reusable'].keys())\n        missing_reuse_attributes = ecs_helpers.list_subtract(['expected', 'top_level'], reuse_attributes)\n        if len(missing_reuse_attributes) > 0:\n            msg = \"Reusable schema {} is missing the following reuse attributes: {}.\\nFound these: {}\".format(\n                schema['field_details']['name'], ', '.join(missing_reuse_attributes), reuse_attributes)\n            raise ValueError(msg)\n\n\ndef schema_assertions_and_warnings(schema):\n    \"\"\"Additional checks on a fleshed out schema\"\"\"\n    single_line_short_description(schema, strict=strict_mode)\n    if 'beta' in schema['field_details']:\n        single_line_beta_description(schema, strict=strict_mode)\n    if 'reusable' in schema['schema_details']:\n        single_line_short_override_description(schema, strict=strict_mode)\n\n\ndef normalize_reuse_notation(schema):\n    \"\"\"\n    Replace single word reuse shorthands from the schema YAMLs with the explicit {at: , as:} notation.\n\n    When marking \"user\" as reusable under \"destination\" with the shorthand entry\n    `- destination`, this is expanded to the complete entry\n    `- { \"at\": \"destination\", \"as\": \"user\" }`.\n    The field set is thus nested at `destination.user.*`, with fields such as `destination.user.name`.\n\n    The dictionary notation enables nesting a field set as a different name.\n    An example is nesting \"process\" fields to capture parent process details\n    at `process.parent.*`.\n    The dictionary notation `- { \"at\": \"process\", \"as\": \"parent\" }` will yield\n    fields such as `process.parent.pid`.\n    \"\"\"\n    if 'reusable' not in schema['schema_details']:\n        return\n    schema_name = schema['field_details']['name']\n    reuse_entries = []\n    for reuse_entry in schema['schema_details']['reusable']['expected']:\n        if type(reuse_entry) is dict:  # Already explicit\n            if 'at' in reuse_entry and 'as' in reuse_entry:\n                explicit_entry = reuse_entry\n            else:\n                raise ValueError(\"When specifying reusable expected locations for {} \" +\n                                 \"with the dictionary notation, keys 'as' and 'at' are required. \" +\n                                 \"Got {}.\".format(schema_name, reuse_entry))\n        else:  # Make it explicit\n            explicit_entry = {'at': reuse_entry, 'as': schema_name}\n        explicit_entry['full'] = explicit_entry['at'] + '.' + explicit_entry['as']\n        reuse_entries.append(explicit_entry)\n    schema['schema_details']['reusable']['expected'] = reuse_entries\n\n\n# Field level cleanup\n\n\ndef field_cleanup(field):\n    field_mandatory_attributes(field)\n    if ecs_helpers.is_intermediate(field):\n        return\n    ecs_helpers.dict_clean_string_values(field['field_details'])\n    if 'allowed_values' in field['field_details']:\n        for allowed_value in field['field_details']['allowed_values']:\n            ecs_helpers.dict_clean_string_values(allowed_value)\n    field_defaults(field)\n    field_assertions_and_warnings(field)\n\n\ndef field_defaults(field):\n    field['field_details'].setdefault('short', field['field_details']['description'])\n    field['field_details'].setdefault('normalize', [])\n    field_or_multi_field_datatype_defaults(field['field_details'])\n    if 'multi_fields' in field['field_details']:\n        for mf in field['field_details']['multi_fields']:\n            field_or_multi_field_datatype_defaults(mf)\n            if 'name' not in mf:\n                mf['name'] = mf['type']\n\n\ndef field_or_multi_field_datatype_defaults(field_details):\n    \"\"\"Sets datatype-related defaults on a canonical field or multi-field entries.\"\"\"\n    if field_details['type'] == 'keyword':\n        field_details.setdefault('ignore_above', 1024)\n    if field_details['type'] == 'text':\n        field_details.setdefault('norms', False)\n    # wildcard needs the index param stripped\n    if field_details['type'] == 'wildcard':\n        field_details.pop('index', None)\n    if 'index' in field_details and not field_details['index']:\n        field_details.setdefault('doc_values', False)\n        field_details.pop('ignore_above', None)\n\n\nFIELD_MANDATORY_ATTRIBUTES = ['name', 'description', 'type', 'level']\nACCEPTABLE_FIELD_LEVELS = ['core', 'extended', 'custom']\n\n\ndef field_mandatory_attributes(field):\n    \"\"\"Ensures for the presence of the mandatory field attributes and raises if any are missing\"\"\"\n    if ecs_helpers.is_intermediate(field):\n        return\n    current_field_attributes = sorted(field['field_details'].keys())\n    missing_attributes = ecs_helpers.list_subtract(FIELD_MANDATORY_ATTRIBUTES, current_field_attributes)\n\n    # `alias` fields require a target `path` attribute.\n    if field['field_details'].get('type') == 'alias' and 'path' not in current_field_attributes:\n        missing_attributes.append('path')\n    # `scaled_float` fields require a `scaling_factor` attribute.\n    if field['field_details'].get('type') == 'scaled_float' and 'scaling_factor' not in current_field_attributes:\n        missing_attributes.append('scaling_factor')\n\n    if len(missing_attributes) > 0:\n        msg = \"Field is missing the following mandatory attributes: {}.\\nFound these: {}.\\nField details: {}\"\n        raise ValueError(msg.format(', '.join(missing_attributes),\n                                    current_field_attributes, field))\n\n\ndef field_assertions_and_warnings(field):\n    \"\"\"Additional checks on a fleshed out field\"\"\"\n    if not ecs_helpers.is_intermediate(field):\n        # check short description length if in strict mode\n        single_line_short_description(field, strict=strict_mode)\n        if 'beta' in field['field_details']:\n            single_line_beta_description(field, strict=strict_mode)\n        if 'pattern' in field['field_details']:\n            validate_pattern_regex(field['field_details'], strict=strict_mode)\n        check_example_value(field, strict=strict_mode)\n        if field['field_details']['level'] not in ACCEPTABLE_FIELD_LEVELS:\n            msg = \"Invalid level for field '{}'.\\nValue: {}\\nAcceptable values: {}\".format(\n                field['field_details']['name'], field['field_details']['level'],\n                ACCEPTABLE_FIELD_LEVELS)\n            raise ValueError(msg)\n\n# Common\n\n\nSHORT_LIMIT = 120\n\n\ndef single_line_short_check(short_to_check, short_name):\n    short_length = len(short_to_check)\n    if \"\\n\" in short_to_check or short_length > SHORT_LIMIT:\n        msg = \"Short descriptions must be single line, and under {} characters (current length: {}).\\n\".format(\n            SHORT_LIMIT, short_length)\n        msg += \"Offending field or field set: {}\\nShort description:\\n  {}\".format(\n            short_name,\n            short_to_check)\n        return msg\n    return None\n\n\ndef strict_warning_handler(message, strict):\n    \"\"\"Handles warnings based on --strict mode\"\"\"\n    if strict:\n        raise ValueError(message)\n    else:\n        ecs_helpers.strict_warning(message)\n\n\ndef single_line_short_description(schema_or_field, strict=True):\n    error = single_line_short_check(schema_or_field['field_details']['short'], schema_or_field['field_details']['name'])\n    if error:\n        strict_warning_handler(error, strict)\n\n\ndef single_line_short_override_description(schema_or_field, strict=True):\n    for field in schema_or_field['schema_details']['reusable']['expected']:\n        if not 'short_override' in field:\n            continue\n        error = single_line_short_check(field['short_override'], field['full'])\n        if error:\n            strict_warning_handler(error, strict)\n\n\ndef check_example_value(field, strict=True):\n    \"\"\"\n    Checks if value of the example field is of type list or dict.\n    Fails or warns (depending on strict mode) if so.\n    \"\"\"\n    example_value = field['field_details'].get('example', None)\n    pattern = field['field_details'].get('pattern', None)\n    name = field['field_details']['name']\n\n    if isinstance(example_value, (list, dict)):\n        msg = f\"Example value for field `{name}` contains an object or array which must be quoted to avoid YAML interpretation.\"\n        strict_warning_handler(msg, strict)\n\n    if pattern:\n        # Examples with arrays must be handled\n        if 'array' in field['field_details']['normalize']:\n            # strips unnecessary chars in order to split each example value\n            example_values = example_value.translate(str.maketrans('', '', '\"[] ')).split(',')\n        else:\n            example_values = [example_value]\n\n        for example_value in example_values:\n            match = re.match(pattern, example_value)\n            if not match:\n                msg = f\"Example value for field `{name}` does not match the regex defined in the pattern attribute: `{pattern}`.\"\n                strict_warning_handler(msg, strict)\n\n\ndef single_line_beta_description(schema_or_field, strict=True):\n    if \"\\n\" in schema_or_field['field_details']['beta']:\n        msg = \"Beta descriptions must be single line.\\n\"\n        msg += f\"Offending field or field set: {schema_or_field['field_details']['name']}\"\n        strict_warning_handler(msg, strict)\n\n\ndef validate_pattern_regex(field, strict=True):\n    \"\"\"\n    Validates if field['pattern'] contains a valid regular expression.\n    \"\"\"\n    try:\n        re.compile(field['pattern'])\n    except re.error:\n        msg = \"Pattern value must be a valid regular expression.\\n\"\n        msg += f\"Offending field name: {field['name']}\"\n        strict_warning_handler(msg, strict)\n", "path": "scripts/schema/cleaner.py"}]}