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
|
# -*- coding: utf-8 -*-
# This Source Code Form is subject to the terms of the 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/.
"""
Validates the contents of a Glean ping against the schema.
"""
import functools
import io
import json
from pathlib import Path
import sys
import jsonschema # type: ignore
from . import util
ROOT_DIR = Path(__file__).parent
SCHEMAS_DIR = ROOT_DIR / "schemas"
@functools.lru_cache(maxsize=1)
def _get_ping_schema(schema_url):
contents = util.fetch_remote_url(schema_url)
return json.loads(contents)
def _validate_ping(ins, outs, schema_url):
schema = _get_ping_schema(schema_url)
resolver = util.get_null_resolver(schema)
document = json.load(ins)
validator_class = jsonschema.validators.validator_for(schema)
validator = validator_class(schema, resolver=resolver)
has_error = 0
for error in validator.iter_errors(document):
outs.write("=" * 76)
outs.write("\n")
outs.write(util.format_error("", "", util.pprint_validation_error(error)))
outs.write("\n")
has_error = 1
return has_error
def validate_ping(ins, outs=None, schema_url=None):
"""
Validates the contents of a Glean ping.
:param ins: Input stream or file path to the ping contents to validate
:param outs: Output stream to write errors to. (Defaults to stdout)
:param schema_url: HTTP URL or local filesystem path to Glean ping schema.
Defaults to the current version of the schema in
mozilla-pipeline-schemas.
:rtype: int 1 if any errors occurred, otherwise 0.
"""
if schema_url is None:
raise TypeError("Missing required argument 'schema_url'")
if outs is None:
outs = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
if isinstance(ins, (str, bytes, Path)):
with open(ins, "r", encoding="utf-8") as fd:
return _validate_ping(fd, outs, schema_url=schema_url)
else:
return _validate_ping(ins, outs, schema_url=schema_url)
|
