86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
|
|
from processdock.services import (
|
|
ProcessService,
|
|
ProtectedProcessError,
|
|
format_command,
|
|
humanize_bytes,
|
|
matches_process,
|
|
)
|
|
|
|
|
|
class FormattingTests(unittest.TestCase):
|
|
def test_humanize_bytes(self) -> None:
|
|
self.assertEqual(humanize_bytes(0), "0 B")
|
|
self.assertEqual(humanize_bytes(1024), "1.0 KB")
|
|
self.assertEqual(humanize_bytes(512 * 1024 * 1024), "512.0 MB")
|
|
|
|
def test_format_command_preserves_arguments(self) -> None:
|
|
command = format_command(["python", "demo app.py", "--port", "8080"])
|
|
self.assertIn("python", command)
|
|
self.assertIn("demo app.py", command)
|
|
self.assertIn("8080", command)
|
|
|
|
def test_searches_name_pid_port_path_and_command(self) -> None:
|
|
process = {
|
|
"pid": 9824,
|
|
"name": "java.exe",
|
|
"username": "wang",
|
|
"exe": r"D:\jdk-17\bin\java.exe",
|
|
"command": "java -jar demo.jar",
|
|
"ports": [8080, 9000],
|
|
}
|
|
for query in ("JAVA", "9824", "8080", "jdk-17", "demo.jar", "wang"):
|
|
with self.subTest(query=query):
|
|
self.assertTrue(matches_process(process, query))
|
|
self.assertFalse(matches_process(process, "nginx"))
|
|
|
|
|
|
class LiveServiceTests(unittest.TestCase):
|
|
def test_snapshot_contains_current_process_and_required_fields(self) -> None:
|
|
snapshot = ProcessService().snapshot()
|
|
current = next(item for item in snapshot["processes"] if item["pid"] == os.getpid())
|
|
required = {
|
|
"pid",
|
|
"name",
|
|
"ports",
|
|
"cpu_percent",
|
|
"memory_bytes",
|
|
"username",
|
|
"status_label",
|
|
"exe",
|
|
"command",
|
|
"create_time_text",
|
|
"ppid",
|
|
}
|
|
self.assertTrue(required.issubset(current))
|
|
self.assertIn("process_count", snapshot["summary"])
|
|
|
|
def test_service_refuses_to_terminate_itself(self) -> None:
|
|
service = ProcessService()
|
|
with self.assertRaises(ProtectedProcessError):
|
|
service.terminate(os.getpid(), force=True)
|
|
|
|
def test_force_terminates_owned_child_process(self) -> None:
|
|
child = subprocess.Popen(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
try:
|
|
result = ProcessService().terminate(child.pid, force=True)
|
|
self.assertTrue(result.success)
|
|
self.assertFalse(result.still_running)
|
|
finally:
|
|
if child.poll() is None:
|
|
child.kill()
|
|
child.wait()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|