Source code
Revision control
Copy as Markdown
Other Tools
# mypy: allow-untyped-defs
import sys
from collections import OrderedDict, defaultdict
from unittest import mock
from mozlog.structuredlog import StructuredLogger
from mozlog.formatters import TbplFormatter
from mozlog.handlers import StreamHandler
from .. import stability, wptrunner
def test_is_inconsistent():
assert stability.is_inconsistent({"PASS": 10}, 10) is False
assert stability.is_inconsistent({"PASS": 9}, 10) is True
assert stability.is_inconsistent({"PASS": 9, "FAIL": 1}, 10) is True
assert stability.is_inconsistent({"PASS": 8, "FAIL": 1}, 10) is True
def test_find_slow_status():
assert stability.find_slow_status({
"longest_duration": {"TIMEOUT": 10},
"timeout": 10}) is None
assert stability.find_slow_status({
"longest_duration": {"CRASH": 10},
"timeout": 10}) is None
assert stability.find_slow_status({
"longest_duration": {"ERROR": 10},
"timeout": 10}) is None
assert stability.find_slow_status({
"longest_duration": {"PASS": 1},
"timeout": 10}) is None
assert stability.find_slow_status({
"longest_duration": {"PASS": 81},
"timeout": 100}) == "PASS"
assert stability.find_slow_status({
"longest_duration": {"TIMEOUT": 10, "FAIL": 81},
"timeout": 100}) == "FAIL"
assert stability.find_slow_status({
"longest_duration": {"SKIP": 0}}) is None
def test_get_steps():
logger = None
steps = stability.get_steps(logger, 0, 0, [])
assert len(steps) == 0
steps = stability.get_steps(logger, 0, 0, [{}])
assert len(steps) == 0
repeat_loop = 1
flag_name = 'flag'
flag_value = 'y'
steps = stability.get_steps(logger, repeat_loop, 0, [
{flag_name: flag_value}])
assert len(steps) == 1
assert steps[0][0] == "Running tests in a loop %d times with flags %s=%s" % (
repeat_loop, flag_name, flag_value)
repeat_loop = 0
repeat_restart = 1
flag_name = 'flag'
flag_value = 'n'
steps = stability.get_steps(logger, repeat_loop, repeat_restart, [
{flag_name: flag_value}])
assert len(steps) == 1
assert steps[0][0] == "Running tests in a loop with restarts %d times with flags %s=%s" % (
repeat_restart, flag_name, flag_value)
repeat_loop = 10
repeat_restart = 5
steps = stability.get_steps(logger, repeat_loop, repeat_restart, [{}])
assert len(steps) == 2
assert steps[0][0] == "Running tests in a loop %d times" % repeat_loop
assert steps[1][0] == (
"Running tests in a loop with restarts %d times" % repeat_restart)
def test_log_handler():
handler = stability.LogHandler()
data = OrderedDict()
data["test"] = "test_name"
test = handler.find_or_create_test(data)
assert test["subtests"] == OrderedDict()
assert test["status"] == defaultdict(int)
assert test["longest_duration"] == defaultdict(float)
assert test == handler.find_or_create_test(data)
start_time = 100
data["time"] = start_time
handler.test_start(data)
assert test["start_time"] == start_time
data["subtest"] = "subtest_name"
subtest = handler.find_or_create_subtest(data)
assert subtest["status"] == defaultdict(int)
assert subtest["messages"] == set()
assert subtest == handler.find_or_create_subtest(data)
data["status"] = 0
assert subtest["status"][data["status"]] == 0
handler.test_status(data)
assert subtest["status"][data["status"]] == 1
handler.test_status(data)
assert subtest["status"][data["status"]] == 2
data["status"] = 1
assert subtest["status"][data["status"]] == 0
message = "test message"
data["message"] = message
handler.test_status(data)
assert subtest["status"][data["status"]] == 1
assert len(subtest["messages"]) == 1
assert message in subtest["messages"]
test_duration = 10
data["time"] = data["time"] + test_duration
handler.test_end(data)
assert test["longest_duration"][data["status"]] == test_duration
assert "timeout" not in test
data["test2"] = "test_name_2"
timeout = 5
data["extra"] = {}
data["extra"]["test_timeout"] = timeout
handler.test_start(data)
handler.test_end(data)
assert test["timeout"] == timeout * 1000
def test_err_string():
assert stability.err_string(
{'OK': 1, 'FAIL': 1}, 1) == "**Duplicate subtest name**"
assert stability.err_string(
{'OK': 2, 'FAIL': 1}, 2) == "**Duplicate subtest name**"
assert stability.err_string({'SKIP': 1}, 0) == "Duplicate subtest name"
assert stability.err_string(
{'SKIP': 1, 'OK': 1}, 1) == "Duplicate subtest name"
assert stability.err_string(
{'FAIL': 1}, 2) == "**FAIL: 1/2, MISSING: 1/2**"
assert stability.err_string(
{'FAIL': 1, 'OK': 1}, 3) == "**FAIL: 1/3, OK: 1/3, MISSING: 1/3**"
assert stability.err_string(
{'OK': 1, 'FAIL': 1}, 2) == "**FAIL: 1/2, OK: 1/2**"
assert stability.err_string(
{'OK': 2, 'FAIL': 1, 'SKIP': 1}, 4) == "FAIL: 1/4, OK: 2/4, SKIP: 1/4"
assert stability.err_string(
{'FAIL': 1, 'SKIP': 1, 'OK': 2}, 4) == "FAIL: 1/4, OK: 2/4, SKIP: 1/4"
def test_check_stability_iterations():
logger = StructuredLogger("test-stability")
logger.add_handler(StreamHandler(sys.stdout, TbplFormatter()))
kwargs = {"verify_log_full": False}
def mock_run_tests(**kwargs):
repeats = kwargs.get("repeat", 1)
for _ in range(repeats):
logger.suite_start(tests=[], name="test")
for _ in range(kwargs.get("rerun", 1)):
logger.test_start("/example/test.html")
logger.test_status("/example/test.html", subtest="test1", status="PASS")
logger.test_end("/example/test.html", status="OK")
logger.suite_end()
status = wptrunner.TestStatus()
status.total_tests = 1
status.repeated_runs = repeats
status.expected_repeated_runs = repeats
return (None, status)
# Don't actually load wptrunner, because that will end up starting a browser
# which we don't want to do in this test.
with mock.patch("wptrunner.stability.wptrunner.run_tests") as mock_run:
mock_run.side_effect = mock_run_tests
assert stability.check_stability(logger,
repeat_loop=10,
repeat_restart=5,
chaos_mode=False,
output_results=False,
**kwargs) is None