mozilla_schema_generator.common_ping
1# -*- coding: utf-8 -*- 2 3# This Source Code Form is subject to the terms of the Mozilla Public 4# License, v. 2.0. If a copy of the MPL was not distributed with this 5# file, You can obtain one at http://mozilla.org/MPL/2.0/. 6 7import json 8from typing import List 9 10from .generic_ping import GenericPing 11from .probes import MainProbe 12from .schema import Schema 13from .utils import prepend_properties 14 15 16class CommonPing(GenericPing): 17 18 # Only includes probes that have been available at some point past 19 # this version. 20 # ONLY DECREMENT, or the schema will change in an incompatible way! 21 MIN_FX_VERSION = 30 22 23 env_url = ( 24 "https://raw.githubusercontent.com/mozilla-services/mozilla-pipeline-schemas" 25 "/{branch}/templates/include/telemetry/environment.1.schema.json" 26 ) 27 probes_url = GenericPing.probe_info_base_url + "/firefox/all/main/all_probes" 28 29 def __init__(self, schema_url, **kwargs): 30 super().__init__(schema_url, self.env_url, self.probes_url, **kwargs) 31 32 def get_schema(self): 33 schema = super().get_schema() 34 35 try: 36 schema.get(prepend_properties(("environment",))) 37 return self._update_env(schema) 38 except KeyError: 39 return schema 40 41 def _update_env(self, schema): 42 integer = {"type": "integer"} 43 string = {"type": "string"} 44 string_map = {"type": "object", "additionalProperties": string} 45 46 def with_description(dtype: dict, comment: str) -> dict: 47 """Add a description to the types defined above.""" 48 return {**dtype, **dict(description=comment)} 49 50 if schema.property_exists(prepend_properties(("environment", "addons"))): 51 active_addons = prepend_properties( 52 ("environment", "addons", "activeAddons") 53 ) + ("additionalProperties", "properties") 54 schema.set_schema_elem(active_addons + ("foreignInstall",), integer) 55 schema.set_schema_elem(active_addons + ("version",), string) 56 schema.set_schema_elem(active_addons + ("userDisabled",), integer) 57 schema.set_schema_elem( 58 prepend_properties( 59 ("environment", "addons", "theme", "foreignInstall") 60 ), 61 integer, 62 ) 63 schema.set_schema_elem( 64 prepend_properties(("environment", "addons", "activeGMPlugins")) 65 + ("additionalProperties", "properties", "applyBackgroundUpdates"), 66 with_description( 67 integer, 68 "Cast into an integer via mozilla-schema-generator. See bug 1611027.", 69 ), 70 ) 71 72 user_prefs = prepend_properties(("environment", "settings", "userPrefs")) 73 if schema.property_exists(user_prefs): 74 desc = ( 75 "User preferences - limited to an allowlist defined in " 76 "`toolkit/components/telemetry/app/TelemetryEnvironment.jsm`" 77 ) 78 schema.set_schema_elem(user_prefs, with_description(string_map, desc)) 79 80 return schema 81 82 def get_env(self): 83 env_property = json.loads("{" + self._get_json_str(self.env_url) + "}") 84 env = {"type": "object", "properties": env_property} 85 86 return self._update_env(Schema(env)) 87 88 def get_probes(self) -> List[MainProbe]: 89 probes = self._get_json(self.probes_url) 90 91 filtered = { 92 pname: pdef 93 for pname, pdef in probes.items() 94 if "nightly" in pdef["first_added"] 95 } 96 97 # This will be made much better with PEP 572 98 main_probes = [MainProbe(_id, defn) for _id, defn in filtered.items()] 99 return [ 100 p 101 for p in main_probes 102 if int(p.definition["versions"]["last"]) > self.MIN_FX_VERSION 103 ]
17class CommonPing(GenericPing): 18 19 # Only includes probes that have been available at some point past 20 # this version. 21 # ONLY DECREMENT, or the schema will change in an incompatible way! 22 MIN_FX_VERSION = 30 23 24 env_url = ( 25 "https://raw.githubusercontent.com/mozilla-services/mozilla-pipeline-schemas" 26 "/{branch}/templates/include/telemetry/environment.1.schema.json" 27 ) 28 probes_url = GenericPing.probe_info_base_url + "/firefox/all/main/all_probes" 29 30 def __init__(self, schema_url, **kwargs): 31 super().__init__(schema_url, self.env_url, self.probes_url, **kwargs) 32 33 def get_schema(self): 34 schema = super().get_schema() 35 36 try: 37 schema.get(prepend_properties(("environment",))) 38 return self._update_env(schema) 39 except KeyError: 40 return schema 41 42 def _update_env(self, schema): 43 integer = {"type": "integer"} 44 string = {"type": "string"} 45 string_map = {"type": "object", "additionalProperties": string} 46 47 def with_description(dtype: dict, comment: str) -> dict: 48 """Add a description to the types defined above.""" 49 return {**dtype, **dict(description=comment)} 50 51 if schema.property_exists(prepend_properties(("environment", "addons"))): 52 active_addons = prepend_properties( 53 ("environment", "addons", "activeAddons") 54 ) + ("additionalProperties", "properties") 55 schema.set_schema_elem(active_addons + ("foreignInstall",), integer) 56 schema.set_schema_elem(active_addons + ("version",), string) 57 schema.set_schema_elem(active_addons + ("userDisabled",), integer) 58 schema.set_schema_elem( 59 prepend_properties( 60 ("environment", "addons", "theme", "foreignInstall") 61 ), 62 integer, 63 ) 64 schema.set_schema_elem( 65 prepend_properties(("environment", "addons", "activeGMPlugins")) 66 + ("additionalProperties", "properties", "applyBackgroundUpdates"), 67 with_description( 68 integer, 69 "Cast into an integer via mozilla-schema-generator. See bug 1611027.", 70 ), 71 ) 72 73 user_prefs = prepend_properties(("environment", "settings", "userPrefs")) 74 if schema.property_exists(user_prefs): 75 desc = ( 76 "User preferences - limited to an allowlist defined in " 77 "`toolkit/components/telemetry/app/TelemetryEnvironment.jsm`" 78 ) 79 schema.set_schema_elem(user_prefs, with_description(string_map, desc)) 80 81 return schema 82 83 def get_env(self): 84 env_property = json.loads("{" + self._get_json_str(self.env_url) + "}") 85 env = {"type": "object", "properties": env_property} 86 87 return self._update_env(Schema(env)) 88 89 def get_probes(self) -> List[MainProbe]: 90 probes = self._get_json(self.probes_url) 91 92 filtered = { 93 pname: pdef 94 for pname, pdef in probes.items() 95 if "nightly" in pdef["first_added"] 96 } 97 98 # This will be made much better with PEP 572 99 main_probes = [MainProbe(_id, defn) for _id, defn in filtered.items()] 100 return [ 101 p 102 for p in main_probes 103 if int(p.definition["versions"]["last"]) > self.MIN_FX_VERSION 104 ]
89 def get_probes(self) -> List[MainProbe]: 90 probes = self._get_json(self.probes_url) 91 92 filtered = { 93 pname: pdef 94 for pname, pdef in probes.items() 95 if "nightly" in pdef["first_added"] 96 } 97 98 # This will be made much better with PEP 572 99 main_probes = [MainProbe(_id, defn) for _id, defn in filtered.items()] 100 return [ 101 p 102 for p in main_probes 103 if int(p.definition["versions"]["last"]) > self.MIN_FX_VERSION 104 ]