"""`main.py sandbox check` 探测函数单元测试。 mock subprocess,验: - daemon 不可达 / image 缺 / network 缺 / uid 错配的各种分支 - detect_fs_quota 对 xfs/ext4/zfs/btrfs/其他 + prjquota mount option 的判断 - 汇总 exit code:全 ok / 仅 warn / 有 err """ from __future__ import annotations import sys import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from core.sandbox.check import ( check_docker_daemon, check_image_present, check_host_uid_alignment, detect_fs_quota, run_sandbox_check, ) def _mk_run(returns): """构造 `_run` 替身:按调用次序返 (rc, stdout, stderr) 列表里的元素。""" iter_ret = iter(returns) def fake_run(argv, timeout=10): return next(iter_ret) return fake_run class TestDaemonCheck(unittest.TestCase): def test_daemon_ok(self): with patch("core.sandbox.check._run", _mk_run([(0, "24.0.7", "")])): self.assertTrue(check_docker_daemon()) def test_daemon_cli_missing(self): with patch("core.sandbox.check._run", _mk_run([(127, "", "docker not found in PATH")])): self.assertFalse(check_docker_daemon()) def test_daemon_permission_denied(self): with patch( "core.sandbox.check._run", _mk_run([(1, "", "Got permission denied while trying to connect")]), ): self.assertFalse(check_docker_daemon()) class TestImageCheck(unittest.TestCase): def test_image_present(self): with patch("core.sandbox.check._run", _mk_run([(0, "[...]", "")])): self.assertTrue(check_image_present()) def test_image_missing(self): with patch("core.sandbox.check._run", _mk_run([(1, "", "No such image")])): self.assertFalse(check_image_present()) class TestHostUidAlignment(unittest.TestCase): def test_uid_aligned(self): if sys.platform == "win32": self.skipTest("getuid not on Windows") import os host_uid = os.getuid() # type: ignore[attr-defined] with patch( "core.sandbox.check._run", _mk_run([(0, str(host_uid), "")]), ): self.assertTrue(check_host_uid_alignment()) def test_uid_mismatch(self): if sys.platform == "win32": self.skipTest("getuid not on Windows") import os bad = os.getuid() + 1 # type: ignore[attr-defined] with patch("core.sandbox.check._run", _mk_run([(0, str(bad), "")])): self.assertFalse(check_host_uid_alignment()) def test_image_not_built_yet(self): # docker run 失败 → warn 不 err with patch( "core.sandbox.check._run", _mk_run([(125, "", "Unable to find image")]), ): self.assertTrue(check_host_uid_alignment()) def test_skipped_on_windows(self): with patch("core.sandbox.check.sys") as mock_sys, \ patch("core.sandbox.check._run", _mk_run([(0, "1000", "")])): mock_sys.platform = "win32" self.assertTrue(check_host_uid_alignment()) class TestDetectFsQuota(unittest.TestCase): """detect_fs_quota:不依赖 print,纯返 (level, msg) 便于断言。""" def test_xfs_with_prjquota(self): with patch("core.sandbox.check._run", _mk_run([(0, "xfs rw,relatime,prjquota,attr2", "")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/opt/zcbot/workspace/users")) self.assertEqual(level, "ok") self.assertIn("xfs with prjquota", msg) def test_xfs_without_prjquota(self): with patch("core.sandbox.check._run", _mk_run([(0, "xfs rw,relatime,attr2", "")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/opt")) self.assertEqual(level, "warn") self.assertIn("NO prjquota", msg) def test_ext4_with_project_quota(self): with patch("core.sandbox.check._run", _mk_run([(0, "ext4 rw,prjquota", "")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/opt")) self.assertEqual(level, "ok") self.assertIn("ext4 with project quota", msg) def test_zfs(self): with patch("core.sandbox.check._run", _mk_run([(0, "zfs rw,xattr,noacl", "")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/tank/zcbot")) self.assertEqual(level, "ok") self.assertIn("zfs", msg) def test_btrfs_warns(self): with patch("core.sandbox.check._run", _mk_run([(0, "btrfs rw,relatime", "")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/opt")) self.assertEqual(level, "warn") self.assertIn("btrfs", msg) def test_tmpfs_warns(self): with patch("core.sandbox.check._run", _mk_run([(0, "tmpfs rw", "")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/tmp")) self.assertEqual(level, "warn") def test_findmnt_missing(self): with patch("core.sandbox.check._run", _mk_run([(127, "", "findmnt not found in PATH")])), \ patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "linux" level, msg = detect_fs_quota(Path("/opt")) self.assertEqual(level, "warn") self.assertIn("findmnt", msg) def test_windows_skipped(self): with patch("core.sandbox.check.sys") as mock_sys: mock_sys.platform = "win32" level, msg = detect_fs_quota(Path("C:/")) self.assertEqual(level, "warn") self.assertIn("Windows", msg) class TestSummaryExitCode(unittest.TestCase): """run_sandbox_check 汇总:err → exit 1,全 ok / 仅 warn → exit 0。""" def test_all_ok_exits_zero(self): with patch("core.sandbox.check.check_docker_daemon", return_value=True), \ patch("core.sandbox.check.check_image_present", return_value=True), \ patch("core.sandbox.check.check_network_present", return_value=True), \ patch("core.sandbox.check.check_host_uid_alignment", return_value=True), \ patch("core.sandbox.check.check_fs_quota_capable", return_value=True): rc = run_sandbox_check() self.assertEqual(rc, 0) def test_any_err_exits_one(self): with patch("core.sandbox.check.check_docker_daemon", return_value=False), \ patch("core.sandbox.check.check_image_present", return_value=True), \ patch("core.sandbox.check.check_network_present", return_value=True), \ patch("core.sandbox.check.check_host_uid_alignment", return_value=True), \ patch("core.sandbox.check.check_fs_quota_capable", return_value=True): rc = run_sandbox_check() self.assertEqual(rc, 1) if __name__ == "__main__": unittest.main()