| 1 | from __future__ import annotations
|
| 2 |
|
| 3 | import sys
|
| 4 | import unittest
|
| 5 | from pathlib import Path
|
| 6 |
|
| 7 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 8 |
|
| 9 | from oakbench.runner import (
|
| 10 | mount_point,
|
| 11 | parse_diskutil_filesystem_type,
|
| 12 | runner_class,
|
| 13 | runner_fields,
|
| 14 | runner_id,
|
| 15 | stable_runner_fields,
|
| 16 | stamp_row,
|
| 17 | )
|
| 18 |
|
| 19 |
|
| 20 | class RunnerProfileTest(unittest.TestCase):
|
| 21 | def test_runner_id_is_stable_for_stable_fields(self) -> None:
|
| 22 | profile = {
|
| 23 | "system": "Linux",
|
| 24 | "release": "6.8",
|
| 25 | "machine": "x86_64",
|
| 26 | "processor": "x86_64",
|
| 27 | "cpu_model": "Test CPU",
|
| 28 | "cpu_count_logical": 8,
|
| 29 | "filesystem_type": "ext2/ext3",
|
| 30 | "probe_path": "/tmp/a",
|
| 31 | }
|
| 32 | changed_probe_path = dict(profile, probe_path="/tmp/b")
|
| 33 | self.assertEqual(stable_runner_fields(profile), stable_runner_fields(changed_probe_path))
|
| 34 | self.assertEqual(runner_id(profile), runner_id(changed_probe_path))
|
| 35 |
|
| 36 | def test_runner_class_first_match_then_default(self) -> None:
|
| 37 | config = {
|
| 38 | "classes": [
|
| 39 | {"id": "linux-test", "match": {"system": "Linux", "machine": "x86_64"}},
|
| 40 | {"id": "unknown", "default": True},
|
| 41 | ]
|
| 42 | }
|
| 43 | self.assertEqual(runner_class({"system": "Linux", "machine": "x86_64"}, config), "linux-test")
|
| 44 | self.assertEqual(runner_class({"system": "Plan9", "machine": "mips"}, config), "unknown")
|
| 45 |
|
| 46 | def test_runner_fields_stamp_rows_additively(self) -> None:
|
| 47 | fields = runner_fields(
|
| 48 | {
|
| 49 | "system": "Linux",
|
| 50 | "release": "6.8",
|
| 51 | "machine": "x86_64",
|
| 52 | "processor": "x86_64",
|
| 53 | "cpu_model": "Test CPU",
|
| 54 | "cpu_count_logical": 8,
|
| 55 | "filesystem_type": "ext4",
|
| 56 | },
|
| 57 | {"classes": [{"id": "unknown", "default": True}]},
|
| 58 | )
|
| 59 | row = {"operation": "status.clean"}
|
| 60 | stamped = stamp_row(row, fields)
|
| 61 | self.assertEqual(row, {"operation": "status.clean"})
|
| 62 | self.assertIn("runner_id", stamped)
|
| 63 | self.assertIn("runner_profile", stamped)
|
| 64 |
|
| 65 | def test_diskutil_parser_returns_stable_filesystem_type_only(self) -> None:
|
| 66 | output = """
|
| 67 | Device Identifier: disk3s1s1
|
| 68 | Volume Free Space: 123456789 Bytes
|
| 69 | File System Personality: APFS
|
| 70 | Type (Bundle): apfs
|
| 71 | """
|
| 72 | self.assertEqual(parse_diskutil_filesystem_type(output), "APFS")
|
| 73 | self.assertIsNone(parse_diskutil_filesystem_type("Volume Free Space: 123\n"))
|
| 74 |
|
| 75 | def test_mount_point_resolves_regular_path_to_mount_root(self) -> None:
|
| 76 | point = mount_point(Path.cwd() / "tests" / "test_runner.py")
|
| 77 | self.assertTrue(point.is_dir())
|
| 78 | self.assertTrue(Path.cwd().resolve().is_relative_to(point) or point == Path("/"))
|
| 79 |
|
| 80 |
|
| 81 | if __name__ == "__main__":
|
| 82 | unittest.main()
|