mirror of
https://github.com/MobileGL-Dev/MobileGL
synced 2026-09-08 04:08:32 +09:00
402 lines
17 KiB
Python
402 lines
17 KiB
Python
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import run_cts_windows as runner
|
|
|
|
|
|
def qpa_closed(case: str, status: str = "Pass") -> str:
|
|
return (
|
|
f"#beginTestCaseResult {case}\n"
|
|
f'<Result StatusCode="{status}">ok</Result>\n'
|
|
"#endTestCaseResult\n"
|
|
)
|
|
|
|
|
|
def command_path(command, option):
|
|
prefix = option + "="
|
|
return Path(next(value[len(prefix) :] for value in command if value.startswith(prefix)))
|
|
|
|
|
|
class AtomicWriteTests(unittest.TestCase):
|
|
def test_access_denied_retries_then_replace_succeeds(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
target = Path(temporary) / "remaining.txt"
|
|
real_replace = runner.os.replace
|
|
attempts = 0
|
|
|
|
def flaky_replace(source, destination):
|
|
nonlocal attempts
|
|
attempts += 1
|
|
if attempts == 1:
|
|
raise PermissionError(13, "temporarily denied", str(destination))
|
|
if attempts == 2:
|
|
error = OSError("temporary WinError 5")
|
|
error.winerror = 5
|
|
raise error
|
|
real_replace(source, destination)
|
|
|
|
with mock.patch.object(runner.os, "replace", side_effect=flaky_replace), mock.patch.object(
|
|
runner.time, "sleep"
|
|
) as sleep:
|
|
runner.atomic_write_text(target, "case.a\n")
|
|
|
|
self.assertEqual(3, attempts)
|
|
self.assertEqual("case.a\n", target.read_text(encoding="utf-8"))
|
|
self.assertEqual(2, sleep.call_count)
|
|
self.assertEqual(
|
|
[
|
|
mock.call(runner.ATOMIC_REPLACE_INITIAL_BACKOFF_SECONDS),
|
|
mock.call(runner.ATOMIC_REPLACE_INITIAL_BACKOFF_SECONDS * 2),
|
|
],
|
|
sleep.call_args_list,
|
|
)
|
|
self.assertEqual([], list(target.parent.glob(".remaining.txt.*.tmp")))
|
|
|
|
def test_permanent_access_denied_stops_after_bounded_attempts(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
target = Path(temporary) / "remaining.txt"
|
|
|
|
def always_denied(_source, _destination):
|
|
error = OSError("persistent WinError 5")
|
|
error.winerror = 5
|
|
raise error
|
|
|
|
with mock.patch.object(
|
|
runner.os, "replace", side_effect=always_denied
|
|
) as replace, mock.patch.object(runner.time, "sleep") as sleep:
|
|
with self.assertRaises(OSError) as raised:
|
|
runner.atomic_write_text(target, "case.a\n")
|
|
|
|
self.assertEqual(5, raised.exception.winerror)
|
|
self.assertEqual(runner.ATOMIC_REPLACE_ATTEMPTS, replace.call_count)
|
|
self.assertEqual(runner.ATOMIC_REPLACE_ATTEMPTS - 1, sleep.call_count)
|
|
self.assertFalse(target.exists())
|
|
self.assertEqual([], list(target.parent.glob(".remaining.txt.*.tmp")))
|
|
|
|
def test_non_access_error_is_not_retried(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
target = Path(temporary) / "remaining.txt"
|
|
error = OSError(28, "disk full")
|
|
with mock.patch.object(
|
|
runner.os, "replace", side_effect=error
|
|
) as replace, mock.patch.object(runner.time, "sleep") as sleep:
|
|
with self.assertRaises(OSError):
|
|
runner.atomic_write_text(target, "case.a\n")
|
|
|
|
self.assertEqual(1, replace.call_count)
|
|
sleep.assert_not_called()
|
|
|
|
|
|
class QpaParsingTests(unittest.TestCase):
|
|
def test_terminate_is_a_completed_result(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "chunk0000.qpa"
|
|
path.write_text(
|
|
"#beginTestCaseResult KHR-GL30.a\n"
|
|
"#terminateTestCaseResult Crash\n"
|
|
"#beginTestCaseResult KHR-GL30.b\n",
|
|
encoding="utf-8",
|
|
)
|
|
progress = runner.scan_qpa(path)
|
|
self.assertEqual(["KHR-GL30.a"], progress.recorded)
|
|
self.assertEqual("KHR-GL30.b", progress.in_flight)
|
|
self.assertEqual(2, progress.begin_count)
|
|
|
|
def test_end_without_result_is_not_accounted(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "chunk0000.qpa"
|
|
path.write_text(
|
|
"#beginTestCaseResult KHR-GL46.incomplete\n"
|
|
"#endTestCaseResult\n"
|
|
+ qpa_closed("KHR-GL46.complete"),
|
|
encoding="utf-8",
|
|
)
|
|
progress = runner.scan_qpa(path)
|
|
self.assertEqual(["KHR-GL46.complete"], progress.recorded)
|
|
self.assertIsNone(progress.in_flight)
|
|
|
|
def test_result_written_before_truncated_eof_is_recovered(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "chunk0000.qpa"
|
|
path.write_text(
|
|
"#beginTestCaseResult KHR-GL46.complete\n"
|
|
'<Result StatusCode="Pass">ok</Result>\n',
|
|
encoding="utf-8",
|
|
)
|
|
progress = runner.scan_qpa(path)
|
|
self.assertEqual(["KHR-GL46.complete"], progress.recorded)
|
|
self.assertIsNone(progress.in_flight)
|
|
|
|
|
|
class RunIdentityTests(unittest.TestCase):
|
|
def test_non_object_run_state_is_a_controlled_error(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
outdir = Path(temporary)
|
|
(outdir / "run_state.json").write_text("null\n", encoding="utf-8")
|
|
with self.assertRaises(runner.RunnerError):
|
|
runner.check_run_identity(outdir, "DirectVulkan", ["case.a"])
|
|
|
|
def test_controller_identity_prevents_mixed_invocations(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
outdir = Path(temporary)
|
|
runner.check_run_identity(
|
|
outdir, "DirectVulkan", ["case.a"], invocation_identity="identity-a"
|
|
)
|
|
runner.check_run_identity(
|
|
outdir, "DirectVulkan", ["case.a"], invocation_identity="identity-a"
|
|
)
|
|
with self.assertRaises(runner.RunnerError):
|
|
runner.check_run_identity(
|
|
outdir, "DirectVulkan", ["case.a"], invocation_identity="identity-b"
|
|
)
|
|
|
|
def test_controller_identity_cannot_be_downgraded_by_omission(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
outdir = Path(temporary)
|
|
runner.check_run_identity(
|
|
outdir, "DirectVulkan", ["case.a"], invocation_identity="identity-a"
|
|
)
|
|
with self.assertRaises(runner.RunnerError):
|
|
runner.check_run_identity(outdir, "DirectVulkan", ["case.a"])
|
|
|
|
def test_legacy_artifacts_require_explicit_adoption(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
outdir = Path(temporary)
|
|
(outdir / "chunk0000.qpa").write_text(
|
|
qpa_closed("case.a"), encoding="utf-8"
|
|
)
|
|
with self.assertRaises(runner.RunnerError):
|
|
runner.check_run_identity(
|
|
outdir, "DirectVulkan", ["case.a"], invocation_identity="identity-a"
|
|
)
|
|
self.assertFalse((outdir / "run_state.json").exists())
|
|
|
|
runner.check_run_identity(
|
|
outdir,
|
|
"DirectVulkan",
|
|
["case.a"],
|
|
invocation_identity="identity-a",
|
|
adopt_legacy=True,
|
|
)
|
|
state = runner.json.loads(
|
|
(outdir / "run_state.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertTrue(state["adopted_legacy"])
|
|
|
|
def test_foreign_nested_qpa_and_skipped_sidecar_are_legacy_evidence(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
outdir = Path(temporary)
|
|
nested = outdir / "old"
|
|
nested.mkdir()
|
|
qpa = nested / "legacy.qpa"
|
|
qpa.write_text(qpa_closed("case.a"), encoding="utf-8")
|
|
skipped = outdir / "skipped.txt"
|
|
skipped.write_text("case.b\n", encoding="utf-8")
|
|
|
|
self.assertEqual(
|
|
{qpa, skipped}, set(runner.recovery_artifacts(outdir))
|
|
)
|
|
with self.assertRaises(runner.RunnerError):
|
|
runner.check_run_identity(
|
|
outdir, "DirectVulkan", ["case.a", "case.b"]
|
|
)
|
|
recorded, crashed, hung = runner.recover_results(
|
|
outdir, {"case.a", "case.b"}
|
|
)
|
|
self.assertEqual({"case.a"}, recorded)
|
|
self.assertEqual(set(), crashed)
|
|
self.assertEqual(set(), hung)
|
|
|
|
def test_non_object_or_non_string_meta_classification_is_ignored(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
outdir = Path(temporary)
|
|
(outdir / "chunk0000.meta.json").write_text("null\n", encoding="utf-8")
|
|
(outdir / "chunk0001.meta.json").write_text("[]\n", encoding="utf-8")
|
|
(outdir / "chunk0002.meta.json").write_text(
|
|
'{"classified_case": [], "classification": "Crash"}\n', encoding="utf-8"
|
|
)
|
|
self.assertEqual(
|
|
(set(), set()), runner.load_meta_classifications(outdir, {"case.a"})
|
|
)
|
|
|
|
|
|
class RunnerRecoveryTests(unittest.TestCase):
|
|
def run_args(self, root: Path, caselist: Path, outdir: Path, *extra: str):
|
|
return [
|
|
"--exe",
|
|
sys.executable,
|
|
"--workdir",
|
|
str(root),
|
|
"--caselist",
|
|
str(caselist),
|
|
"--outdir",
|
|
str(outdir),
|
|
"--backend",
|
|
"DirectVulkan",
|
|
*extra,
|
|
]
|
|
|
|
def test_crash_tail_is_quarantined_and_next_chunk_resumes(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
caselist = root / "cases.txt"
|
|
outdir = root / "results"
|
|
caselist.write_text("KHR-GL30.a\nKHR-GL30.b\nKHR-GL30.c\n", encoding="utf-8")
|
|
seen_remaining = []
|
|
|
|
def fake_run(command, workdir, environment, qpa_path, stdout_path, stderr_path, *timeouts):
|
|
del workdir, timeouts
|
|
seen_remaining.append(
|
|
command_path(command, "--deqp-caselist-file")
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
)
|
|
stdout_path.write_text("fake stdout\n", encoding="utf-8")
|
|
stderr_path.write_text("fake stderr\n", encoding="utf-8")
|
|
self.assertEqual("DirectVulkan", environment["MOBILEGL_BACKEND_TYPE"])
|
|
if len(seen_remaining) == 1:
|
|
qpa_path.write_text(
|
|
qpa_closed("KHR-GL30.a") + "#beginTestCaseResult KHR-GL30.b\n",
|
|
encoding="utf-8",
|
|
)
|
|
return runner.ProcessOutcome(0xC0000005, 0.1)
|
|
qpa_path.write_text(qpa_closed("KHR-GL30.c"), encoding="utf-8")
|
|
return runner.ProcessOutcome(0, 0.1)
|
|
|
|
with mock.patch.object(runner, "run_process", side_effect=fake_run):
|
|
result = runner.main(self.run_args(root, caselist, outdir))
|
|
|
|
self.assertEqual(0, result)
|
|
self.assertEqual(
|
|
[
|
|
["KHR-GL30.a", "KHR-GL30.b", "KHR-GL30.c"],
|
|
["KHR-GL30.c"],
|
|
],
|
|
seen_remaining,
|
|
)
|
|
self.assertEqual("KHR-GL30.b\n", (outdir / "crashed.txt").read_text(encoding="utf-8"))
|
|
self.assertEqual("", (outdir / "hung.txt").read_text(encoding="utf-8"))
|
|
self.assertEqual("", (outdir / "unrun.txt").read_text(encoding="utf-8"))
|
|
|
|
def test_existing_qpa_and_sidecar_are_recovered(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
caselist = root / "cases.txt"
|
|
outdir = root / "results"
|
|
outdir.mkdir()
|
|
caselist.write_text("KHR-GL31.a\nKHR-GL31.b\nKHR-GL31.c\n", encoding="utf-8")
|
|
(outdir / "chunk0000.qpa").write_text(qpa_closed("KHR-GL31.a"), encoding="utf-8")
|
|
(outdir / "crashed.txt").write_text("KHR-GL31.b\n", encoding="utf-8")
|
|
seen_remaining = []
|
|
|
|
def fake_run(command, workdir, environment, qpa_path, stdout_path, stderr_path, *timeouts):
|
|
del workdir, environment, stdout_path, stderr_path, timeouts
|
|
seen_remaining.extend(
|
|
command_path(command, "--deqp-caselist-file")
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
)
|
|
qpa_path.write_text(qpa_closed("KHR-GL31.c"), encoding="utf-8")
|
|
return runner.ProcessOutcome(0, 0.1)
|
|
|
|
with mock.patch.object(runner, "run_process", side_effect=fake_run):
|
|
result = runner.main(
|
|
self.run_args(root, caselist, outdir, "--adopt-legacy")
|
|
)
|
|
|
|
self.assertEqual(0, result)
|
|
self.assertEqual(["KHR-GL31.c"], seen_remaining)
|
|
self.assertTrue((outdir / "chunk0001.qpa").is_file())
|
|
|
|
def test_repeated_no_output_aborts_without_false_case_blame(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
caselist = root / "cases.txt"
|
|
outdir = root / "results"
|
|
caselist.write_text("KHR-GL32.a\nKHR-GL32.b\n", encoding="utf-8")
|
|
seen_remaining = []
|
|
|
|
def fake_run(command, workdir, environment, qpa_path, stdout_path, stderr_path, *timeouts):
|
|
del workdir, environment, stdout_path, stderr_path, timeouts
|
|
seen_remaining.append(
|
|
command_path(command, "--deqp-caselist-file")
|
|
.read_text(encoding="utf-8")
|
|
.splitlines()
|
|
)
|
|
qpa_path.write_text("#sessionInfo releaseName fake\n", encoding="utf-8")
|
|
return runner.ProcessOutcome(
|
|
1, 0.1, timed_out=True, timeout_reason="qpa-idle"
|
|
)
|
|
|
|
with mock.patch.object(runner, "run_process", side_effect=fake_run):
|
|
result = runner.main(
|
|
self.run_args(root, caselist, outdir, "--max-empty-streak", "2")
|
|
)
|
|
|
|
self.assertEqual(4, result)
|
|
self.assertEqual(
|
|
[["KHR-GL32.a", "KHR-GL32.b"], ["KHR-GL32.a", "KHR-GL32.b"]],
|
|
seen_remaining,
|
|
)
|
|
self.assertEqual("", (outdir / "crashed.txt").read_text(encoding="utf-8"))
|
|
self.assertEqual("", (outdir / "hung.txt").read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
"KHR-GL32.a\nKHR-GL32.b\n",
|
|
(outdir / "unrun.txt").read_text(encoding="utf-8"),
|
|
)
|
|
|
|
|
|
class ProcessTimeoutTests(unittest.TestCase):
|
|
def test_qpa_activity_prevents_idle_timeout(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
qpa = root / "active.qpa"
|
|
helper = (
|
|
"import pathlib,sys,time\n"
|
|
"path=pathlib.Path(sys.argv[1])\n"
|
|
"for size in range(1, 9):\n"
|
|
" path.write_text('x' * size, encoding='utf-8')\n"
|
|
" time.sleep(0.08)\n"
|
|
)
|
|
outcome = runner.run_process(
|
|
[sys.executable, "-c", helper, str(qpa)],
|
|
root,
|
|
dict(runner.os.environ),
|
|
qpa,
|
|
root / "stdout.log",
|
|
root / "stderr.log",
|
|
idle_timeout=0.2,
|
|
max_round_seconds=0,
|
|
poll_seconds=0.03,
|
|
)
|
|
self.assertFalse(outcome.timed_out)
|
|
self.assertEqual(0, outcome.returncode)
|
|
|
|
def test_idle_timeout_really_stops_process(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
started = time.monotonic()
|
|
outcome = runner.run_process(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
root,
|
|
dict(runner.os.environ),
|
|
root / "never-created.qpa",
|
|
root / "stdout.log",
|
|
root / "stderr.log",
|
|
idle_timeout=0.2,
|
|
max_round_seconds=0,
|
|
poll_seconds=0.05,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
self.assertTrue(outcome.timed_out)
|
|
self.assertEqual("qpa-idle", outcome.timeout_reason)
|
|
self.assertLess(elapsed, 10)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|