147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
# 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
|
|
"""
|
|
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(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()
|