39 lines
1.5 KiB
Python
39 lines
1.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
from cmbuyer_client.core.errors import SingleInstanceError
|
||
|
|
from cmbuyer_client.localstate.single_instance import NamedMutex
|
||
|
|
|
||
|
|
|
||
|
|
@unittest.skipUnless(os.name == "nt", "named mutex 仅在 Windows 验证")
|
||
|
|
class NamedMutexTests(unittest.TestCase):
|
||
|
|
def test_second_process_for_same_database_is_rejected(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as directory:
|
||
|
|
database = Path(directory) / "state.sqlite3"
|
||
|
|
first = NamedMutex(database)
|
||
|
|
try:
|
||
|
|
with self.assertRaises(SingleInstanceError):
|
||
|
|
NamedMutex(database)
|
||
|
|
code = (
|
||
|
|
"from pathlib import Path; "
|
||
|
|
"from cmbuyer_client.localstate.single_instance import NamedMutex; "
|
||
|
|
"from cmbuyer_client.core.errors import SingleInstanceError; "
|
||
|
|
f"p=Path({str(database)!r}); "
|
||
|
|
"\ntry:\n NamedMutex(p)\nexcept SingleInstanceError:\n raise SystemExit(17)\nraise SystemExit(0)"
|
||
|
|
)
|
||
|
|
environment = dict(os.environ)
|
||
|
|
environment["PYTHONPATH"] = str(Path(__file__).resolve().parents[2] / "src")
|
||
|
|
result = subprocess.run([sys.executable, "-c", code], env=environment, check=False)
|
||
|
|
self.assertEqual(result.returncode, 17)
|
||
|
|
finally:
|
||
|
|
first.close()
|
||
|
|
|
||
|
|
with NamedMutex(database):
|
||
|
|
pass
|