wehub-resource-sync
c6af9e284a
Tests / catch-all (windows-latest) (push) Has been cancelled
Tests / jvm (macos-latest) (push) Has been cancelled
Tests / jvm (ubuntu-latest) (push) Has been cancelled
Tests / jvm (windows-latest) (push) Has been cancelled
Tests / native (macos-latest) (push) Has been cancelled
Tests / native (ubuntu-latest) (push) Has been cancelled
Tests / native (windows-latest) (push) Has been cancelled
Tests / niche (ubuntu-latest) (push) Has been cancelled
Tests / other-langs (macos-latest) (push) Has been cancelled
Tests / other-langs (ubuntu-latest) (push) Has been cancelled
Tests / other-langs (windows-latest) (push) Has been cancelled
Tests / catch-all (macos-latest) (push) Has been cancelled
Tests / catch-all (ubuntu-latest) (push) Has been cancelled
Docs Build / build (push) Has been cancelled
Docs Build / deploy (push) Has been cancelled
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Codespell / Check for spelling errors (push) Has been cancelled
Build and Push Docker Images / build-and-push (push) Has been cancelled
269 lines
9.1 KiB
Python
269 lines
9.1 KiB
Python
"""
|
|
Unit tests for the metals_db_utils module.
|
|
|
|
Tests the detection of Metals H2 database status and stale lock handling.
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from solidlsp.util.metals_db_utils import (
|
|
MetalsDbStatus,
|
|
check_metals_db_status,
|
|
cleanup_stale_lock,
|
|
is_metals_process_alive,
|
|
parse_h2_lock_file,
|
|
)
|
|
|
|
|
|
@pytest.mark.scala
|
|
class TestParseH2LockFile:
|
|
"""Tests for parse_h2_lock_file function."""
|
|
|
|
def test_returns_none_when_file_does_not_exist(self, tmp_path: Path) -> None:
|
|
"""Should return None when lock file doesn't exist."""
|
|
lock_path = tmp_path / "nonexistent.lock.db"
|
|
result = parse_h2_lock_file(lock_path)
|
|
assert result is None
|
|
|
|
def test_parses_server_format_lock_file(self, tmp_path: Path) -> None:
|
|
"""Should parse lock file with server:host:port format."""
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
lock_path.write_text("server:localhost:9092\n")
|
|
|
|
result = parse_h2_lock_file(lock_path)
|
|
|
|
assert result is not None
|
|
assert result.port == 9092
|
|
assert result.lock_path == lock_path
|
|
|
|
def test_parses_port_only_format(self, tmp_path: Path) -> None:
|
|
"""Should extract port from content containing a port number."""
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
lock_path.write_text("some content 9123 more content\n")
|
|
|
|
result = parse_h2_lock_file(lock_path)
|
|
|
|
assert result is not None
|
|
assert result.port == 9123
|
|
|
|
def test_parses_pid_format(self, tmp_path: Path) -> None:
|
|
"""Should extract PID from lock file content."""
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
lock_path.write_text("pid=12345\nserver:localhost:9092\n")
|
|
|
|
result = parse_h2_lock_file(lock_path)
|
|
|
|
assert result is not None
|
|
assert result.pid == 12345
|
|
assert result.port == 9092
|
|
|
|
def test_handles_unreadable_file(self, tmp_path: Path) -> None:
|
|
"""Should return None for unreadable files."""
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
lock_path.write_text("content")
|
|
# Make file unreadable (Unix only)
|
|
if os.name != "nt":
|
|
lock_path.chmod(0o000)
|
|
try:
|
|
result = parse_h2_lock_file(lock_path)
|
|
assert result is None
|
|
finally:
|
|
lock_path.chmod(0o644)
|
|
|
|
def test_truncates_raw_content(self, tmp_path: Path) -> None:
|
|
"""Should truncate raw_content to 200 chars."""
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
long_content = "x" * 500
|
|
lock_path.write_text(long_content)
|
|
|
|
result = parse_h2_lock_file(lock_path)
|
|
|
|
assert result is not None
|
|
assert len(result.raw_content) == 200
|
|
|
|
|
|
@pytest.mark.scala
|
|
class TestIsMetalsProcessAlive:
|
|
"""Tests for is_metals_process_alive function."""
|
|
|
|
def test_returns_false_for_nonexistent_process(self) -> None:
|
|
"""Should return False for a PID that doesn't exist."""
|
|
# Use a very high PID that's unlikely to exist
|
|
result = is_metals_process_alive(999999999)
|
|
assert result is False
|
|
|
|
def test_returns_true_for_metals_process(self) -> None:
|
|
"""Should return True for a running Metals process."""
|
|
import psutil
|
|
|
|
with patch.object(psutil, "Process") as mock_process_class:
|
|
mock_proc = MagicMock()
|
|
mock_proc.is_running.return_value = True
|
|
mock_proc.cmdline.return_value = [
|
|
"java",
|
|
"-Dmetals.client=vscode",
|
|
"-jar",
|
|
"metals.jar",
|
|
]
|
|
mock_process_class.return_value = mock_proc
|
|
|
|
result = is_metals_process_alive(12345)
|
|
|
|
assert result is True
|
|
|
|
def test_returns_false_for_non_metals_java_process(self) -> None:
|
|
"""Should return False for a Java process that isn't Metals."""
|
|
import psutil
|
|
|
|
with patch.object(psutil, "Process") as mock_process_class:
|
|
mock_proc = MagicMock()
|
|
mock_proc.is_running.return_value = True
|
|
mock_proc.cmdline.return_value = [
|
|
"java",
|
|
"-jar",
|
|
"some-other-app.jar",
|
|
]
|
|
mock_process_class.return_value = mock_proc
|
|
|
|
result = is_metals_process_alive(12345)
|
|
|
|
assert result is False
|
|
|
|
def test_returns_false_for_non_running_process(self) -> None:
|
|
"""Should return False for a process that's not running."""
|
|
import psutil
|
|
|
|
with patch.object(psutil, "Process") as mock_process_class:
|
|
mock_proc = MagicMock()
|
|
mock_proc.is_running.return_value = False
|
|
mock_process_class.return_value = mock_proc
|
|
|
|
result = is_metals_process_alive(12345)
|
|
|
|
assert result is False
|
|
|
|
def test_handles_no_such_process(self) -> None:
|
|
"""Should return False when process doesn't exist."""
|
|
import psutil
|
|
|
|
with patch.object(psutil, "Process") as mock_process_class:
|
|
mock_process_class.side_effect = psutil.NoSuchProcess(12345)
|
|
|
|
result = is_metals_process_alive(12345)
|
|
|
|
assert result is False
|
|
|
|
|
|
@pytest.mark.scala
|
|
class TestCheckMetalsDbStatus:
|
|
"""Tests for check_metals_db_status function."""
|
|
|
|
def test_returns_no_database_when_metals_dir_missing(self, tmp_path: Path) -> None:
|
|
"""Should return NO_DATABASE when .metals directory doesn't exist."""
|
|
status, lock_info = check_metals_db_status(tmp_path)
|
|
|
|
assert status == MetalsDbStatus.NO_DATABASE
|
|
assert lock_info is None
|
|
|
|
def test_returns_no_database_when_db_missing(self, tmp_path: Path) -> None:
|
|
"""Should return NO_DATABASE when database file doesn't exist."""
|
|
metals_dir = tmp_path / ".metals"
|
|
metals_dir.mkdir()
|
|
|
|
status, lock_info = check_metals_db_status(tmp_path)
|
|
|
|
assert status == MetalsDbStatus.NO_DATABASE
|
|
assert lock_info is None
|
|
|
|
def test_returns_no_lock_when_lock_file_missing(self, tmp_path: Path) -> None:
|
|
"""Should return NO_LOCK when database exists but lock doesn't."""
|
|
metals_dir = tmp_path / ".metals"
|
|
metals_dir.mkdir()
|
|
db_path = metals_dir / "metals.mv.db"
|
|
db_path.touch()
|
|
|
|
status, lock_info = check_metals_db_status(tmp_path)
|
|
|
|
assert status == MetalsDbStatus.NO_LOCK
|
|
assert lock_info is None
|
|
|
|
def test_returns_active_instance_when_process_alive(self, tmp_path: Path) -> None:
|
|
"""Should return ACTIVE_INSTANCE when lock holder is running."""
|
|
import solidlsp.util.metals_db_utils as metals_utils
|
|
|
|
metals_dir = tmp_path / ".metals"
|
|
metals_dir.mkdir()
|
|
db_path = metals_dir / "metals.mv.db"
|
|
db_path.touch()
|
|
lock_path = metals_dir / "metals.mv.db.lock.db"
|
|
lock_path.write_text("pid=12345\nserver:localhost:9092\n")
|
|
|
|
with patch.object(metals_utils, "is_metals_process_alive", return_value=True):
|
|
status, lock_info = check_metals_db_status(tmp_path)
|
|
|
|
assert status == MetalsDbStatus.ACTIVE_INSTANCE
|
|
assert lock_info is not None
|
|
assert lock_info.is_stale is False
|
|
|
|
def test_returns_stale_lock_when_process_dead(self, tmp_path: Path) -> None:
|
|
"""Should return STALE_LOCK when lock holder is not running."""
|
|
import solidlsp.util.metals_db_utils as metals_utils
|
|
|
|
metals_dir = tmp_path / ".metals"
|
|
metals_dir.mkdir()
|
|
db_path = metals_dir / "metals.mv.db"
|
|
db_path.touch()
|
|
lock_path = metals_dir / "metals.mv.db.lock.db"
|
|
lock_path.write_text("pid=12345\nserver:localhost:9092\n")
|
|
|
|
with patch.object(metals_utils, "is_metals_process_alive", return_value=False):
|
|
status, lock_info = check_metals_db_status(tmp_path)
|
|
|
|
assert status == MetalsDbStatus.STALE_LOCK
|
|
assert lock_info is not None
|
|
assert lock_info.is_stale is True
|
|
|
|
|
|
@pytest.mark.scala
|
|
class TestCleanupStaleLock:
|
|
"""Tests for cleanup_stale_lock function."""
|
|
|
|
def test_removes_lock_file(self, tmp_path: Path) -> None:
|
|
"""Should successfully remove a lock file."""
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
lock_path.touch()
|
|
|
|
result = cleanup_stale_lock(lock_path)
|
|
|
|
assert result is True
|
|
assert not lock_path.exists()
|
|
|
|
def test_returns_true_when_file_already_removed(self, tmp_path: Path) -> None:
|
|
"""Should return True when file doesn't exist."""
|
|
lock_path = tmp_path / "nonexistent.lock.db"
|
|
|
|
result = cleanup_stale_lock(lock_path)
|
|
|
|
assert result is True
|
|
|
|
def test_returns_false_on_permission_error(self, tmp_path: Path) -> None:
|
|
"""Should return False when file can't be removed due to permissions."""
|
|
if os.name == "nt":
|
|
pytest.skip("Permission test not reliable on Windows")
|
|
|
|
lock_path = tmp_path / "metals.mv.db.lock.db"
|
|
lock_path.touch()
|
|
# Make parent directory read-only
|
|
tmp_path.chmod(0o555)
|
|
|
|
try:
|
|
result = cleanup_stale_lock(lock_path)
|
|
assert result is False
|
|
assert lock_path.exists()
|
|
finally:
|
|
tmp_path.chmod(0o755)
|