chendelian
6c75ade463
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 25s
Add precise Chrome/Edge quit by user-data-dir and stale SingletonLock cleanup so publish RPAs can free account-manager windows before launch without killing the default browser.
135 lines
5.5 KiB
Python
135 lines
5.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Profile 精准关闭 / 僵锁清理单元测试。"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from jiangchang_skill_core.rpa.errors import ERR_PROFILE_IN_USE
|
|
from jiangchang_skill_core.rpa.profile_lock import (
|
|
clear_stale_singleton_locks,
|
|
ensure_profile_free,
|
|
force_quit_profile_browsers,
|
|
is_profile_lock_error,
|
|
)
|
|
|
|
|
|
class TestProfileLockHelpers(unittest.TestCase):
|
|
def test_is_profile_lock_error(self) -> None:
|
|
self.assertTrue(
|
|
is_profile_lock_error(
|
|
RuntimeError("User data directory is already in use")
|
|
)
|
|
)
|
|
self.assertTrue(is_profile_lock_error(RuntimeError("SingletonLock")))
|
|
self.assertFalse(is_profile_lock_error(RuntimeError("timeout waiting")))
|
|
|
|
def test_force_quit_rejects_short_path(self) -> None:
|
|
self.assertEqual(force_quit_profile_browsers("C:\\a"), 0)
|
|
|
|
def test_clear_stale_singleton_when_no_pids(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
profile = Path(tmp) / "account-manager" / "profiles" / "acc-1"
|
|
profile.mkdir(parents=True)
|
|
lock = profile / "SingletonLock"
|
|
cookie = profile / "SingletonCookie"
|
|
lock.write_text("deadhost-999999", encoding="utf-8")
|
|
cookie.write_text("x", encoding="utf-8")
|
|
with mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
|
|
return_value=[],
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock._pid_alive",
|
|
return_value=False,
|
|
):
|
|
removed = clear_stale_singleton_locks(str(profile))
|
|
self.assertTrue(removed)
|
|
self.assertFalse(lock.exists())
|
|
self.assertFalse(cookie.exists())
|
|
|
|
def test_clear_stale_skips_when_pids_present(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
profile = Path(tmp) / "profiles" / "acc-2"
|
|
profile.mkdir(parents=True)
|
|
lock = profile / "SingletonLock"
|
|
lock.write_text("host-1", encoding="utf-8")
|
|
with mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
|
|
return_value=[1234],
|
|
):
|
|
removed = clear_stale_singleton_locks(str(profile))
|
|
self.assertFalse(removed)
|
|
self.assertTrue(lock.exists())
|
|
|
|
def test_ensure_profile_free_quits_then_ok(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
profile = Path(tmp) / "profiles" / "acc-long-enough"
|
|
profile.mkdir(parents=True)
|
|
with mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
|
|
side_effect=[[111], []],
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.force_quit_profile_browsers",
|
|
return_value=1,
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.clear_stale_singleton_locks",
|
|
return_value=False,
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.time.sleep",
|
|
):
|
|
result = ensure_profile_free(
|
|
str(profile), wait_sec=0, raise_if_busy=True
|
|
)
|
|
self.assertTrue(result.freed)
|
|
self.assertEqual(result.pids_before, (111,))
|
|
self.assertEqual(result.pids_after, ())
|
|
self.assertEqual(result.killed_ops, 1)
|
|
|
|
def test_ensure_profile_free_raises_when_still_busy(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
profile = Path(tmp) / "profiles" / "acc-still-busy"
|
|
profile.mkdir(parents=True)
|
|
with mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
|
|
return_value=[222],
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.force_quit_profile_browsers",
|
|
return_value=1,
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.clear_stale_singleton_locks",
|
|
return_value=False,
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.time.sleep",
|
|
):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
ensure_profile_free(str(profile), wait_sec=0, raise_if_busy=True)
|
|
self.assertIn(ERR_PROFILE_IN_USE, str(ctx.exception))
|
|
|
|
def test_ensure_profile_free_no_raise_when_requested(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
profile = Path(tmp) / "profiles" / "acc-keep-busy"
|
|
profile.mkdir(parents=True)
|
|
with mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
|
|
return_value=[333],
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.force_quit_profile_browsers",
|
|
return_value=0,
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.clear_stale_singleton_locks",
|
|
return_value=False,
|
|
), mock.patch(
|
|
"jiangchang_skill_core.rpa.profile_lock.time.sleep",
|
|
):
|
|
result = ensure_profile_free(
|
|
str(profile), wait_sec=0, raise_if_busy=False
|
|
)
|
|
self.assertFalse(result.freed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|