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
|
# This Source Code Form is subject to the terms of Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import os
import sys
import unittest
from os import path
import mozunit
import yaml
TELEMETRY_ROOT_PATH = path.abspath(
path.join(path.dirname(__file__), path.pardir, path.pardir)
)
sys.path.append(TELEMETRY_ROOT_PATH)
# The parsers live in a subdirectory of "build_scripts", account for that.
# NOTE: if the parsers are moved, this logic will need to be updated.
sys.path.append(path.join(TELEMETRY_ROOT_PATH, "build_scripts"))
from mozparsers import parse_events
from mozparsers.shared_telemetry_utils import ParserError
def load_event(event):
"""Parse the passed event and return a dictionary
:param event: Event as YAML string
:returns: Parsed Event dictionary
"""
return yaml.safe_load(event)
class TestParser(unittest.TestCase):
def setUp(self):
def mockexit(x):
raise SystemExit(x)
self.oldexit = os._exit
os._exit = mockexit
def tearDown(self):
os._exit = self.oldexit
def test_valid_event_defaults(self):
SAMPLE_EVENT = """
objects: ["object1", "object2"]
bug_numbers: [12345]
notification_emails: ["test01@mozilla.com", "test02@mozilla.com"]
record_in_processes: ["main"]
description: This is a test entry for Telemetry.
products: ["firefox"]
expiry_version: never
"""
name = "test_event"
event = load_event(SAMPLE_EVENT)
evt = parse_events.EventData("CATEGORY", name, event, strict_type_checks=True)
ParserError.exit_func()
self.assertEqual(evt.methods, [name])
self.assertEqual(evt.record_in_processes, ["main"])
self.assertEqual(evt.objects, ["object1", "object2"])
self.assertEqual(evt.products, ["firefox"])
self.assertEqual(evt.operating_systems, ["all"])
self.assertEqual(evt.extra_keys, [])
def test_wrong_collection(self):
SAMPLE_EVENT = """
objects: ["object1", "object2"]
bug_numbers: [12345]
notification_emails: ["test01@mozilla.com", "test02@mozilla.com"]
record_in_processes: ["main"]
description: This is a test entry for Telemetry.
expiry_version: never
products: ["firefox"]
release_channel_collection: none
"""
event = load_event(SAMPLE_EVENT)
parse_events.EventData("CATEGORY", "test_event", event, strict_type_checks=True)
self.assertRaises(SystemExit, ParserError.exit_func)
def test_valid_event_custom(self):
SAMPLE_EVENT = """
methods: ["method1", "method2"]
objects: ["object1", "object2"]
bug_numbers: [12345]
notification_emails: ["test01@mozilla.com", "test02@mozilla.com"]
record_in_processes: ["content"]
description: This is a test entry for Telemetry.
expiry_version: never
extra_keys:
key1: test1
key2: test2
products:
- fennec
operating_systems:
- windows
"""
name = "test_event"
event = load_event(SAMPLE_EVENT)
evt = parse_events.EventData("CATEGORY", name, event, strict_type_checks=True)
ParserError.exit_func()
self.assertEqual(evt.methods, ["method1", "method2"])
self.assertEqual(evt.objects, ["object1", "object2"])
self.assertEqual(evt.record_in_processes, ["content"])
self.assertEqual(evt.products, ["fennec"])
self.assertEqual(evt.operating_systems, ["windows"])
self.assertEqual(sorted(evt.extra_keys), ["key1", "key2"])
def test_absent_products(self):
SAMPLE_EVENT = """
methods: ["method1", "method2"]
objects: ["object1", "object2"]
bug_numbers: [12345]
notification_emails: ["test01@mozilla.com", "test02@mozilla.com"]
record_in_processes: ["content"]
description: This is a test entry for Telemetry.
expiry_version: never
"""
event = load_event(SAMPLE_EVENT)
self.assertRaises(
SystemExit,
lambda: parse_events.EventData(
"CATEGORY", "test_event", event, strict_type_checks=True
),
)
def test_empty_products(self):
SAMPLE_EVENT = """
methods: ["method1", "method2"]
objects: ["object1", "object2"]
bug_numbers: [12345]
notification_emails: ["test01@mozilla.com", "test02@mozilla.com"]
record_in_processes: ["content"]
description: This is a test entry for Telemetry.
products: []
expiry_version: never
"""
event = load_event(SAMPLE_EVENT)
self.assertRaises(
SystemExit,
lambda: parse_events.EventData(
"CATEGORY", "test_event", event, strict_type_checks=True
),
)
if __name__ == "__main__":
mozunit.main()
|