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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
# -*- coding: utf-8 -*-
# Any copyright is dedicated to the Public Domain.
# http://creativecommons.org/publicdomain/zero/1.0/
from pathlib import Path
import io
import json
import pytest
import subprocess
import glean_parser
from glean_parser import translate
from glean_parser import validate_ping
ROOT = Path(__file__).parent
def test_parser_rb_server_ping_file(tmp_path, capsys):
"""Test that no files are generated if ping definition
is provided."""
translate.translate(
[
ROOT / "data" / "server_metrics_with_event.yaml",
ROOT / "data" / "server_pings.yaml",
],
"ruby_server",
tmp_path,
)
captured = capsys.readouterr()
assert all(False for _ in tmp_path.iterdir())
assert (
"Ping definition found. Server-side environment is simplified" in captured.out
)
def test_parser_rb_server_no_event_metrics(tmp_path, capsys):
"""Test that no files are generated if no event metrics."""
translate.translate(
[ROOT / "data" / "server_metrics_no_events_no_pings.yaml"],
"ruby_server",
tmp_path,
)
captured = capsys.readouterr()
assert all(False for _ in tmp_path.iterdir())
assert (
"No event metrics found...at least one event metric is required" in captured.out
)
def test_parser_rb_server_metrics_unsupported_type(tmp_path, capsys):
"""Test that no files are generated with unsupported metric types."""
translate.translate(
[
ROOT / "data" / "ruby_server_metrics_unsupported.yaml",
],
"ruby_server",
tmp_path,
)
captured = capsys.readouterr()
assert "Ignoring unsupported metric type" in captured.out
assert "boolean" in captured.out
def test_parser_rb_server_pings_unsupported_type(tmp_path, capsys):
"""Test that no files are generated with ping types that are not `events`."""
translate.translate(
[
ROOT / "data" / "ruby_server_pings_unsupported.yaml",
],
"ruby_server",
tmp_path,
)
captured = capsys.readouterr()
assert "Non-events ping reference found" in captured.out
assert "Ignoring the tests ping type" in captured.out
def test_parser_rb_server(tmp_path):
"""Test that parser works"""
translate.translate(
[ROOT / "data" / "server_metrics_with_event.yaml"],
"ruby_server",
tmp_path,
)
assert set(x.name for x in tmp_path.iterdir()) == set(["server_events.rb"])
# Make sure string metric made it in
with (tmp_path / "server_events.rb").open("r", encoding="utf-8") as fd:
content = fd.read()
with (ROOT / "data" / "server_events_compare.rb").open(
"r", encoding="utf-8"
) as cd:
compare_raw = cd.read()
compare = compare_raw.format(
current_version=f"glean_parser v{glean_parser.__version__}"
)
assert content == compare
def run_logger(code_dir, import_file, code):
"""
Run the Ruby logger with a mocked logger
that just prints the ping payload to STDOUT.
"""
tmpl_code = ""
with open(ROOT / "test-rb" / "test.rb.tmpl", "r") as fp:
tmpl_code = fp.read()
tmpl_code = tmpl_code.replace("/* IMPORT */", import_file).replace(
"/* CODE */", code
)
with open(code_dir / "test.rb", "w") as fp:
fp.write(tmpl_code)
return subprocess.check_output(["ruby", "test.rb"], cwd=code_dir).decode("utf-8")
@pytest.mark.ruby_dependency
def test_run_logging(tmp_path):
translate.translate(
[
ROOT / "data" / "server_metrics_with_event.yaml",
],
"ruby_server",
tmp_path,
)
code = """
events.backend_object_update.record(
object_type: "type",
object_state: "state",
linking: true,
identifiers_fxa_account_id: nil,
user_agent: "glean-test/1.0",
ip_address: "127.0.0.1"
)
"""
logged_output = run_logger(tmp_path, "server_events.rb", code)
logged_output = json.loads(logged_output)
fields = logged_output["Fields"]
payload = fields["payload"]
assert "glean-server-event" == logged_output["Type"]
assert "glean.test" == fields["document_namespace"]
assert "events" == fields["document_type"]
assert "1" == fields["document_version"]
assert "glean-test/1.0" == fields["user_agent"]
schema_url = (
"https://raw.githubusercontent.com/mozilla-services/"
"mozilla-pipeline-schemas/main/"
"schemas/glean/glean/glean.1.schema.json"
)
input = io.StringIO(payload)
output = io.StringIO()
assert (
validate_ping.validate_ping(input, output, schema_url=schema_url) == 0
), output.getvalue()
|
