zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-08-03 14:27:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: reorganize unit test files into appropriate directories (#13343)

Browse Source 
### What problem does this PR solve?

Move test files from utils/ to their corresponding functional
directories:
- api/db/ for database related tests
- api/utils/ for API utility tests
- rag/utils/ for RAG utility tests

### Type of change

- [x] Refactoring
This commit is contained in:
Liu An
2026-03-04 11:02:56 +08:00
committed by GitHub
parent 33ba955b02
commit 7715bad04e
7 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
test/unit_test/api/utils/test_api_file_utils.py Normal file
View File

@@ -0,0 +1,152 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Unit tests for api.utils.file_utils (filename_type, thumbnail_img, sanitize_path, read_potential_broken_pdf)."""
import pytest
from api.db import FileType
from api.utils.file_utils import (
MAX_BLOB_SIZE_PDF,
MAX_BLOB_SIZE_THUMBNAIL,
GHOSTSCRIPT_TIMEOUT_SEC,
filename_type,
thumbnail_img,
thumbnail,
sanitize_path,
read_potential_broken_pdf,
repair_pdf_with_ghostscript,
)
class TestFilenameType:
"""Edge cases and robustness for filename_type."""
@pytest.mark.parametrize("filename,expected", [
("doc.pdf", FileType.PDF.value),
("a.PDF", FileType.PDF.value),
("x.png", FileType.VISUAL.value),
("file.docx", FileType.DOC.value),
("a/b/c.pdf", FileType.PDF.value),
("path/to/file.txt", FileType.DOC.value),
])
def test_valid_filenames(self, filename, expected):
assert filename_type(filename) == expected
@pytest.mark.parametrize("filename", [
None,
"",
" ",
123,
[],
])
def test_invalid_or_empty_returns_other(self, filename):
assert filename_type(filename) == FileType.OTHER.value
def test_path_with_basename_uses_extension(self):
assert filename_type("folder/subfolder/document.pdf") == FileType.PDF.value
class TestSanitizePath:
"""Edge cases for sanitize_path."""
@pytest.mark.parametrize("raw,expected", [
(None, ""),
("", ""),
(" ", ""),
(42, ""),
("a/b", "a/b"),
("a/../b", "a/b"),
("/leading/", "leading"),
("\\mixed\\path", "mixed/path"),
])
def test_sanitize_cases(self, raw, expected):
assert sanitize_path(raw) == expected
class TestReadPotentialBrokenPdf:
"""Edge cases and robustness for read_potential_broken_pdf."""
def test_none_returns_empty_bytes(self):
assert read_potential_broken_pdf(None) == b""
def test_empty_bytes_returns_as_is(self):
assert read_potential_broken_pdf(b"") == b""
def test_non_len_raises_or_returns_empty(self):
class NoLen:
pass
result = read_potential_broken_pdf(NoLen())
assert result == b""
class TestThumbnailImg:
"""Edge cases for thumbnail_img."""
def test_none_blob_returns_none(self):
assert thumbnail_img("x.pdf", None) is None
def test_none_filename_returns_none(self):
assert thumbnail_img(None, b"fake pdf content") is None
def test_empty_blob_returns_none(self):
assert thumbnail_img("x.pdf", b"") is None
def test_empty_filename_returns_none(self):
assert thumbnail_img("", b"x") is None
def test_oversized_blob_returns_none(self):
huge = b"x" * (MAX_BLOB_SIZE_THUMBNAIL + 1)
assert thumbnail_img("x.pdf", huge) is None
class TestThumbnail:
"""thumbnail() wraps thumbnail_img and returns base64 or empty string."""
def test_none_img_returns_empty_string(self):
assert thumbnail("x.xyz", b"garbage") == ""
def test_valid_img_returns_base64_prefix(self):
from api.constants import IMG_BASE64_PREFIX
result = thumbnail("x.png", b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00\x00\x01\x01\x00\x05\x18\xd8N\x00\x00\x00\x00IEND\xaeB`\x82")
assert result.startswith(IMG_BASE64_PREFIX) or result == ""
class TestRepairPdfWithGhostscript:
"""repair_pdf_with_ghostscript edge cases."""
def test_none_returns_empty_bytes(self):
assert repair_pdf_with_ghostscript(None) == b""
def test_empty_bytes_returns_empty(self):
assert repair_pdf_with_ghostscript(b"") == b""
def test_oversized_returns_original_without_calling_gs(self):
huge = b"%" * (MAX_BLOB_SIZE_PDF + 1)
result = repair_pdf_with_ghostscript(huge)
assert result == huge
class TestConstants:
"""Resource limit constants are positive and reasonable."""
def test_thumbnail_limit_positive(self):
assert MAX_BLOB_SIZE_THUMBNAIL > 0
def test_pdf_limit_positive(self):
assert MAX_BLOB_SIZE_PDF > 0
def test_gs_timeout_positive(self):
assert GHOSTSCRIPT_TIMEOUT_SEC > 0

146
test/unit_test/api/utils/test_health_utils_minio.py Normal file
View File

@@ -0,0 +1,146 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""
Unit tests for MinIO health check (check_minio_alive) and scheme/verify helpers.
Covers SSL/HTTPS and certificate verification (issues #13158, #13159).
"""
from unittest.mock import patch, Mock
class TestMinioSchemeAndVerify:
"""Test _minio_scheme_and_verify helper."""
@patch("api.utils.health_utils.settings")
def test_scheme_http_when_secure_false(self, mock_settings):
mock_settings.MINIO = {"host": "minio:9000", "secure": False}
from api.utils.health_utils import _minio_scheme_and_verify
scheme, verify = _minio_scheme_and_verify()
assert scheme == "http"
assert verify is True
@patch("api.utils.health_utils.settings")
def test_scheme_https_when_secure_true(self, mock_settings):
mock_settings.MINIO = {"host": "minio:9000", "secure": True}
from api.utils.health_utils import _minio_scheme_and_verify
scheme, verify = _minio_scheme_and_verify()
assert scheme == "https"
assert verify is True
@patch("api.utils.health_utils.settings")
def test_scheme_https_when_secure_string_true(self, mock_settings):
mock_settings.MINIO = {"host": "minio:9000", "secure": "true"}
from api.utils.health_utils import _minio_scheme_and_verify
scheme, verify = _minio_scheme_and_verify()
assert scheme == "https"
@patch("api.utils.health_utils.settings")
def test_verify_false_for_self_signed(self, mock_settings):
mock_settings.MINIO = {"host": "minio:9000", "secure": True, "verify": False}
from api.utils.health_utils import _minio_scheme_and_verify
scheme, verify = _minio_scheme_and_verify()
assert scheme == "https"
assert verify is False
@patch("api.utils.health_utils.settings")
def test_verify_string_false(self, mock_settings):
mock_settings.MINIO = {"host": "minio:9000", "verify": "false"}
from api.utils.health_utils import _minio_scheme_and_verify
_, verify = _minio_scheme_and_verify()
assert verify is False
@patch("api.utils.health_utils.settings")
def test_default_verify_true_when_key_missing(self, mock_settings):
mock_settings.MINIO = {"host": "minio:9000"}
from api.utils.health_utils import _minio_scheme_and_verify
_, verify = _minio_scheme_and_verify()
assert verify is True
class TestCheckMinioAlive:
"""Test check_minio_alive with mocked requests and settings."""
@patch("api.utils.health_utils.requests.get")
@patch("api.utils.health_utils.settings")
def test_returns_alive_when_http_200(self, mock_settings, mock_get):
mock_settings.MINIO = {"host": "minio:9000", "secure": False}
mock_response = Mock()
mock_response.status_code = 200
mock_get.return_value = mock_response
from api.utils.health_utils import check_minio_alive
result = check_minio_alive()
assert result["status"] == "alive"
assert "elapsed" in result["message"]
mock_get.assert_called_once()
call_args = mock_get.call_args
assert call_args[0][0] == "http://minio:9000/minio/health/live"
assert call_args[1]["verify"] is True
@patch("api.utils.health_utils.requests.get")
@patch("api.utils.health_utils.settings")
def test_uses_https_when_secure_true(self, mock_settings, mock_get):
mock_settings.MINIO = {"host": "minio:9000", "secure": True}
mock_response = Mock()
mock_response.status_code = 200
mock_get.return_value = mock_response
from api.utils.health_utils import check_minio_alive
check_minio_alive()
call_args = mock_get.call_args
assert call_args[0][0] == "https://minio:9000/minio/health/live"
@patch("api.utils.health_utils.requests.get")
@patch("api.utils.health_utils.settings")
def test_passes_verify_false_for_self_signed(self, mock_settings, mock_get):
mock_settings.MINIO = {"host": "minio:9000", "secure": True, "verify": False}
mock_response = Mock()
mock_response.status_code = 200
mock_get.return_value = mock_response
from api.utils.health_utils import check_minio_alive
check_minio_alive()
call_args = mock_get.call_args
assert call_args[1]["verify"] is False
@patch("api.utils.health_utils.requests.get")
@patch("api.utils.health_utils.settings")
def test_returns_timeout_on_non_200(self, mock_settings, mock_get):
mock_settings.MINIO = {"host": "minio:9000"}
mock_response = Mock()
mock_response.status_code = 503
mock_get.return_value = mock_response
from api.utils.health_utils import check_minio_alive
result = check_minio_alive()
assert result["status"] == "timeout"
@patch("api.utils.health_utils.requests.get")
@patch("api.utils.health_utils.settings")
def test_returns_timeout_on_request_exception(self, mock_settings, mock_get):
mock_settings.MINIO = {"host": "minio:9000"}
mock_get.side_effect = ConnectionError("Connection refused")
from api.utils.health_utils import check_minio_alive
result = check_minio_alive()
assert result["status"] == "timeout"
assert "error" in result["message"]
@patch("api.utils.health_utils.requests.get")
@patch("api.utils.health_utils.settings")
def test_request_uses_timeout(self, mock_settings, mock_get):
mock_settings.MINIO = {"host": "minio:9000"}
mock_response = Mock()
mock_response.status_code = 200
mock_get.return_value = mock_response
from api.utils.health_utils import check_minio_alive
check_minio_alive()
call_args = mock_get.call_args
assert call_args[1]["timeout"] == 10

412
test/unit_test/api/utils/test_oceanbase_health.py Normal file
View File

@@ -0,0 +1,412 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""
Unit tests for OceanBase health check and performance monitoring functionality.
"""
import inspect
import os
import types
import pytest
from unittest.mock import Mock, patch
from api.utils.health_utils import get_oceanbase_status, check_oceanbase_health
class TestOceanBaseHealthCheck:
"""Test cases for OceanBase health check functionality."""
@patch('api.utils.health_utils.OBConnection')
@patch.dict(os.environ, {'DOC_ENGINE': 'oceanbase'})
def test_get_oceanbase_status_success(self, mock_ob_class):
"""Test successful OceanBase status retrieval."""
# Setup mock
mock_ob_connection = Mock()
mock_ob_connection.uri = "localhost:2881"
mock_ob_connection.health.return_value = {
"uri": "localhost:2881",
"version_comment": "OceanBase 4.3.5.1",
"status": "healthy",
"connection": "connected"
}
mock_ob_connection.get_performance_metrics.return_value = {
"connection": "connected",
"latency_ms": 5.2,
"storage_used": "1.2MB",
"storage_total": "100GB",
"query_per_second": 150,
"slow_queries": 2,
"active_connections": 10,
"max_connections": 300
}
mock_ob_class.return_value = mock_ob_connection
# Execute
result = get_oceanbase_status()
# Assert
assert result["status"] == "alive"
assert "message" in result
assert "health" in result["message"]
assert "performance" in result["message"]
assert result["message"]["health"]["status"] == "healthy"
assert result["message"]["performance"]["latency_ms"] == 5.2
@patch.dict(os.environ, {'DOC_ENGINE': 'elasticsearch'})
def test_get_oceanbase_status_not_configured(self):
"""Test OceanBase status when not configured."""
with pytest.raises(Exception) as exc_info:
get_oceanbase_status()
assert "OceanBase is not in use" in str(exc_info.value)
@patch('api.utils.health_utils.OBConnection')
@patch.dict(os.environ, {'DOC_ENGINE': 'oceanbase'})
def test_get_oceanbase_status_connection_error(self, mock_ob_class):
"""Test OceanBase status when connection fails."""
mock_ob_class.side_effect = Exception("Connection failed")
result = get_oceanbase_status()
assert result["status"] == "timeout"
assert "error" in result["message"]
@patch('api.utils.health_utils.OBConnection')
@patch.dict(os.environ, {'DOC_ENGINE': 'oceanbase'})
def test_check_oceanbase_health_healthy(self, mock_ob_class):
"""Test OceanBase health check returns healthy status."""
mock_ob_connection = Mock()
mock_ob_connection.health.return_value = {
"uri": "localhost:2881",
"version_comment": "OceanBase 4.3.5.1",
"status": "healthy",
"connection": "connected"
}
mock_ob_connection.get_performance_metrics.return_value = {
"connection": "connected",
"latency_ms": 5.2,
"storage_used": "1.2MB",
"storage_total": "100GB",
"query_per_second": 150,
"slow_queries": 0,
"active_connections": 10,
"max_connections": 300
}
mock_ob_class.return_value = mock_ob_connection
result = check_oceanbase_health()
assert result["status"] == "healthy"
assert result["details"]["connection"] == "connected"
assert result["details"]["latency_ms"] == 5.2
assert result["details"]["query_per_second"] == 150
@patch('api.utils.health_utils.OBConnection')
@patch.dict(os.environ, {'DOC_ENGINE': 'oceanbase'})
def test_check_oceanbase_health_degraded(self, mock_ob_class):
"""Test OceanBase health check returns degraded status for high latency."""
mock_ob_connection = Mock()
mock_ob_connection.health.return_value = {
"uri": "localhost:2881",
"version_comment": "OceanBase 4.3.5.1",
"status": "healthy",
"connection": "connected"
}
mock_ob_connection.get_performance_metrics.return_value = {
"connection": "connected",
"latency_ms": 1500.0, # High latency > 1000ms
"storage_used": "1.2MB",
"storage_total": "100GB",
"query_per_second": 50,
"slow_queries": 5,
"active_connections": 10,
"max_connections": 300
}
mock_ob_class.return_value = mock_ob_connection
result = check_oceanbase_health()
assert result["status"] == "degraded"
assert result["details"]["latency_ms"] == 1500.0
@patch('api.utils.health_utils.OBConnection')
@patch.dict(os.environ, {'DOC_ENGINE': 'oceanbase'})
def test_check_oceanbase_health_unhealthy(self, mock_ob_class):
"""Test OceanBase health check returns unhealthy status."""
mock_ob_connection = Mock()
mock_ob_connection.health.return_value = {
"uri": "localhost:2881",
"status": "unhealthy",
"connection": "disconnected",
"error": "Connection timeout"
}
mock_ob_connection.get_performance_metrics.return_value = {
"connection": "disconnected",
"error": "Connection timeout"
}
mock_ob_class.return_value = mock_ob_connection
result = check_oceanbase_health()
assert result["status"] == "unhealthy"
assert result["details"]["connection"] == "disconnected"
assert "error" in result["details"]
@patch.dict(os.environ, {'DOC_ENGINE': 'elasticsearch'})
def test_check_oceanbase_health_not_configured(self):
"""Test OceanBase health check when not configured."""
result = check_oceanbase_health()
assert result["status"] == "not_configured"
assert result["details"]["connection"] == "not_configured"
assert "not configured" in result["details"]["message"].lower()
class TestOBConnectionPerformanceMetrics:
"""Test cases for OBConnection performance metrics methods."""
def _create_mock_connection(self):
"""Create a mock OBConnection with actual methods."""
# Create a simple object and bind the real methods to it
class MockConn:
pass
conn = MockConn()
# Get the actual class from the singleton wrapper's closure
from rag.utils import ob_conn
# OBConnection is wrapped by @singleton decorator, so it's a function
# The original class is stored in the closure of the singleton function
# Find the class by checking all closure cells
ob_connection_class = None
if hasattr(ob_conn.OBConnection, '__closure__') and ob_conn.OBConnection.__closure__:
for cell in ob_conn.OBConnection.__closure__:
cell_value = cell.cell_contents
if inspect.isclass(cell_value):
ob_connection_class = cell_value
break
if ob_connection_class is None:
raise ValueError("Could not find OBConnection class in closure")
# Bind the actual methods to our mock object
conn.get_performance_metrics = types.MethodType(ob_connection_class.get_performance_metrics, conn)
conn._get_storage_info = types.MethodType(ob_connection_class._get_storage_info, conn)
conn._get_connection_pool_stats = types.MethodType(ob_connection_class._get_connection_pool_stats, conn)
conn._get_slow_query_count = types.MethodType(ob_connection_class._get_slow_query_count, conn)
conn._estimate_qps = types.MethodType(ob_connection_class._estimate_qps, conn)
return conn
def test_get_performance_metrics_success(self):
"""Test successful retrieval of performance metrics."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.uri = "localhost:2881"
conn.db_name = "test"
# Mock client methods - create separate mock results for each call
mock_result1 = Mock()
mock_result1.fetchone.return_value = (1,)
mock_result2 = Mock()
mock_result2.fetchone.return_value = (100.5,)
mock_result3 = Mock()
mock_result3.fetchone.return_value = (100.0,)
mock_result4 = Mock()
mock_result4.fetchall.return_value = [
(1, 'user', 'host', 'db', 'Query', 0, 'executing', 'SELECT 1')
]
mock_result4.fetchone.return_value = ('max_connections', '300')
mock_result5 = Mock()
mock_result5.fetchone.return_value = (0,)
mock_result6 = Mock()
mock_result6.fetchone.return_value = (5,)
# Setup side_effect to return different mocks for different queries
def sql_side_effect(query):
if "SELECT 1" in query:
return mock_result1
elif "information_schema.tables" in query:
return mock_result2
elif "__all_disk_stat" in query:
return mock_result3
elif "SHOW PROCESSLIST" in query:
return mock_result4
elif "SHOW VARIABLES LIKE 'max_connections'" in query:
return mock_result4
elif "information_schema.processlist" in query and "time >" in query:
return mock_result5
elif "information_schema.processlist" in query and "COUNT" in query:
return mock_result6
return Mock()
mock_client.perform_raw_text_sql.side_effect = sql_side_effect
mock_client.pool_size = 300
# Mock logger
import logging
conn.logger = logging.getLogger('test')
result = conn.get_performance_metrics()
assert result["connection"] == "connected"
assert result["latency_ms"] >= 0
assert "storage_used" in result
assert "storage_total" in result
def test_get_performance_metrics_connection_error(self):
"""Test performance metrics when connection fails."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.uri = "localhost:2881"
conn.logger = Mock()
mock_client.perform_raw_text_sql.side_effect = Exception("Connection failed")
result = conn.get_performance_metrics()
assert result["connection"] == "disconnected"
assert "error" in result
def test_get_storage_info_success(self):
"""Test successful retrieval of storage information."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.db_name = "test"
conn.logger = Mock()
mock_result1 = Mock()
mock_result1.fetchone.return_value = (100.5,)
mock_result2 = Mock()
mock_result2.fetchone.return_value = (100.0,)
def sql_side_effect(query):
if "information_schema.tables" in query:
return mock_result1
elif "__all_disk_stat" in query:
return mock_result2
return Mock()
mock_client.perform_raw_text_sql.side_effect = sql_side_effect
result = conn._get_storage_info()
assert "storage_used" in result
assert "storage_total" in result
assert "MB" in result["storage_used"]
def test_get_storage_info_fallback(self):
"""Test storage info with fallback when total space unavailable."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.db_name = "test"
conn.logger = Mock()
# First query succeeds, second fails
def side_effect(query):
if "information_schema.tables" in query:
mock_result = Mock()
mock_result.fetchone.return_value = (100.5,)
return mock_result
else:
raise Exception("Table not found")
mock_client.perform_raw_text_sql.side_effect = side_effect
result = conn._get_storage_info()
assert "storage_used" in result
assert "storage_total" in result
def test_get_connection_pool_stats(self):
"""Test retrieval of connection pool statistics."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.logger = Mock()
mock_client.pool_size = 300
mock_result1 = Mock()
mock_result1.fetchall.return_value = [
(1, 'user', 'host', 'db', 'Query', 0, 'executing', 'SELECT 1'),
(2, 'user', 'host', 'db', 'Sleep', 10, None, None)
]
mock_result2 = Mock()
mock_result2.fetchone.return_value = ('max_connections', '300')
def sql_side_effect(query):
if "SHOW PROCESSLIST" in query:
return mock_result1
elif "SHOW VARIABLES LIKE 'max_connections'" in query:
return mock_result2
return Mock()
mock_client.perform_raw_text_sql.side_effect = sql_side_effect
result = conn._get_connection_pool_stats()
assert "active_connections" in result
assert "max_connections" in result
assert result["active_connections"] >= 0
def test_get_slow_query_count(self):
"""Test retrieval of slow query count."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.logger = Mock()
mock_result = Mock()
mock_result.fetchone.return_value = (5,)
mock_client.perform_raw_text_sql.return_value = mock_result
result = conn._get_slow_query_count(threshold_seconds=1)
assert isinstance(result, int)
assert result >= 0
def test_estimate_qps(self):
"""Test QPS estimation."""
# Create mock connection with actual methods
conn = self._create_mock_connection()
mock_client = Mock()
conn.client = mock_client
conn.logger = Mock()
mock_result = Mock()
mock_result.fetchone.return_value = (10,)
mock_client.perform_raw_text_sql.return_value = mock_result
result = conn._estimate_qps()
assert isinstance(result, int)
assert result >= 0
if __name__ == "__main__":
pytest.main([__file__, "-v"])