Test Case Update (#515)

* UI/UX improvements

Signed-off-by: Mohan Lakshmaiah <[email protected]>

* Test case update

Signed-off-by: Mohan Lakshmaiah <[email protected]>

---------

Signed-off-by: Mohan Lakshmaiah <[email protected]>

Author: MohanLaksh

tests/unit/mcpgateway/utils/test_error_formatter.py

@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+"""
+Full-coverage unit tests for **mcpgateway.utils.error_formatter**
+
+Running:
+
+    pytest -q --cov=mcpgateway.utils.error_formatter --cov-report=term-missing
+
+Should show **100 %** statement coverage for the target module.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Author: Mihai Criveti
+"""
+
+import pytest
+from unittest.mock import Mock
+from pydantic import BaseModel, ValidationError, field_validator
+from sqlalchemy.exc import DatabaseError, IntegrityError
+
+from mcpgateway.utils.error_formatter import ErrorFormatter
+
+class NameModel(BaseModel):
+    name: str
+
+    @field_validator('name')
+    def validate_name(cls, v):
+        if not v.startswith('A'):
+            raise ValueError('Tool name must start with a letter')
+        if len(v) > 255:
+            raise ValueError('Tool name exceeds maximum length')
+        return v
+
+class UrlModel(BaseModel):
+    url: str
+
+    @field_validator('url')
+    def validate_url(cls, v):
+        if not v.startswith('http'):
+            raise ValueError('Tool URL must start with http')
+        return v
+
+class PathModel(BaseModel):
+    path: str
+
+    @field_validator('path')
+    def validate_path(cls, v):
+        if '..' in v:
+            raise ValueError('cannot contain directory traversal')
+        return v
+
+class ContentModel(BaseModel):
+    content: str
+
+    @field_validator('content')
+    def validate_content(cls, v):
+        if '<' in v and '>' in v:
+            raise ValueError('contains HTML tags')
+        return v
+
+def test_format_validation_error_letter_requirement():
+    with pytest.raises(ValidationError) as exc:
+        NameModel(name="Bobby")
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert result["message"] == "Validation failed"
+    assert result["success"] is False
+    assert result["details"][0]["field"] == "name"
+    assert "must start with a letter" in result["details"][0]["message"]
+
+def test_format_validation_error_length():
+    with pytest.raises(ValidationError) as exc:
+        NameModel(name="A" * 300)
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert "too long" in result["details"][0]["message"]
+
+def test_format_validation_error_url():
+    with pytest.raises(ValidationError) as exc:
+        UrlModel(url="ftp://example.com")
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert "valid HTTP" in result["details"][0]["message"]
+
+def test_format_validation_error_directory_traversal():
+    with pytest.raises(ValidationError) as exc:
+        PathModel(path="../etc/passwd")
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert "invalid characters" in result["details"][0]["message"]
+
+def test_format_validation_error_html_injection():
+    with pytest.raises(ValidationError) as exc:
+        ContentModel(content="<script>alert(1)</script>")
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert "cannot contain HTML" in result["details"][0]["message"]
+
+def test_format_validation_error_fallback():
+    class CustomModel(BaseModel):
+        custom: str
+
+        @field_validator('custom')
+        def validate_custom(cls, v):
+            raise ValueError('Some unknown error')
+    with pytest.raises(ValidationError) as exc:
+        CustomModel(custom="foo")
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert result["details"][0]["message"] == "Invalid custom"
+
+def test_format_validation_error_multiple_fields():
+    class MultiModel(BaseModel):
+        name: str
+        url: str
+
+        @field_validator('name')
+        def validate_name(cls, v):
+            if len(v) > 255:
+                raise ValueError('Tool name exceeds maximum length')
+            return v
+
+        @field_validator('url')
+        def validate_url(cls, v):
+            if not v.startswith('http'):
+                raise ValueError('Tool URL must start with http')
+            return v
+
+    with pytest.raises(ValidationError) as exc:
+        MultiModel(name="A" * 300, url="ftp://bad")
+    result = ErrorFormatter.format_validation_error(exc.value)
+    assert len(result["details"]) == 2
+    messages = [d["message"] for d in result["details"]]
+    assert any("too long" in m for m in messages)
+    assert any("valid HTTP" in m for m in messages)
+
+def test_get_user_message_all_patterns():
+    # Directly test _get_user_message for all mappings and fallback
+    assert "must start with a letter" in ErrorFormatter._get_user_message("name", "Tool name must start with a letter")
+    assert "too long" in ErrorFormatter._get_user_message("description", "Tool name exceeds maximum length")
+    assert "valid HTTP" in ErrorFormatter._get_user_message("endpoint", "Tool URL must start with http")
+    assert "invalid characters" in ErrorFormatter._get_user_message("path", "cannot contain directory traversal")
+    assert "cannot contain HTML" in ErrorFormatter._get_user_message("content", "contains HTML tags")
+    assert ErrorFormatter._get_user_message("foo", "random error") == "Invalid foo"
+
+def make_mock_integrity_error(msg):
+    mock = Mock(spec=IntegrityError)
+    mock.orig = Mock()
+    mock.orig.__str__ = lambda self=mock.orig: msg
+    return mock
+
+@pytest.mark.parametrize("msg,expected", [
+    ("UNIQUE constraint failed: gateways.url", "A gateway with this URL already exists"),
+    ("UNIQUE constraint failed: gateways.name", "A gateway with this name already exists"),
+    ("UNIQUE constraint failed: tools.name", "A tool with this name already exists"),
+    ("UNIQUE constraint failed: resources.uri", "A resource with this URI already exists"),
+    ("UNIQUE constraint failed: servers.name", "A server with this name already exists"),
+    ("FOREIGN KEY constraint failed", "Referenced item not found"),
+    ("NOT NULL constraint failed", "Required field is missing"),
+    ("CHECK constraint failed: invalid_data", "Validation failed. Please check the input data."),
+])
+def test_format_database_error_integrity_patterns(msg, expected):
+    err = make_mock_integrity_error(msg)
+    result = ErrorFormatter.format_database_error(err)
+    assert result["message"] == expected
+    assert result["success"] is False
+
+def test_format_database_error_generic_integrity():
+    err = make_mock_integrity_error("SOME OTHER ERROR")
+    result = ErrorFormatter.format_database_error(err)
+    assert result["message"].startswith("Unable to complete")
+    assert result["success"] is False
+
+def test_format_database_error_generic_database():
+    mock = Mock(spec=DatabaseError)
+    mock.orig = None
+    result = ErrorFormatter.format_database_error(mock)
+    assert result["message"].startswith("Unable to complete")
+    assert result["success"] is False
+
+def test_format_database_error_no_orig():
+    # Simulate error without .orig attribute
+    class DummyError(Exception):
+        pass
+    dummy = DummyError("fail")
+    result = ErrorFormatter.format_database_error(dummy)
+    assert result["message"].startswith("Unable to complete")
+    assert result["success"] is False

tests/unit/mcpgateway/utils/test_redis_isready.py

@@ -10,6 +10,8 @@
 # Standard
 import asyncio
 from unittest.mock import patch
+import sys
+import types
 
 # Third-Party
 import pytest
@@ -198,3 +200,111 @@ def from_url(cls, url):
         )
 
     assert mock.attempts == 1
+
+
+def test_importerror_exits(monkeypatch):
+    """If redis is not installed, should exit with code 2."""
+    # Patch sys.modules to simulate ImportError
+    import importlib
+    import builtins
+
+    # Save original import
+    orig_import = builtins.__import__
+
+    def fake_import(name, *args, **kwargs):
+        if name == "redis":
+            raise ImportError("No redis")
+        return orig_import(name, *args, **kwargs)
+
+    builtins.__import__ = fake_import
+    # Patch sys.exit to catch exit code
+    exit_code = {}
+    monkeypatch.setattr(sys, "exit", lambda code: (_ for _ in ()).throw(SystemExit(code)))
+    try:
+        with pytest.raises(SystemExit) as exc:
+            # Use sync=True to hit ImportError path
+            import mcpgateway.utils.redis_isready as redis_isready_mod
+            redis_isready_mod.wait_for_redis_ready(sync=True)
+        assert exc.value.code == 2
+    finally:
+        builtins.__import__ = orig_import
+
+
+def test_logging_config(monkeypatch):
+    """Logger with no handlers triggers basicConfig."""
+    import logging
+    # Use a real Logger instance and clear its handlers
+    dummy_logger = logging.getLogger("dummy_logger_for_test_logging_config")
+    dummy_logger.handlers.clear()
+    # Patch Redis to always succeed
+    class DummyRedis:
+        @classmethod
+        def from_url(cls, url): return cls()
+        def ping(self): return True
+    monkeypatch.setattr("redis.Redis", DummyRedis)
+    monkeypatch.setattr("mcpgateway.utils.redis_isready.time.sleep", lambda *_: None)
+    import mcpgateway.utils.redis_isready as redis_isready_mod
+    redis_isready_mod.wait_for_redis_ready(logger=dummy_logger, sync=True)
+
+
+def test_parse_cli_and_main_success(monkeypatch):
+    """Test CLI parse and main() success path (exit 0)."""
+    import mcpgateway.utils.redis_isready as redis_isready_mod
+    # Patch sys.argv and sys.exit
+    monkeypatch.setattr(sys, "argv", ["redis_isready.py", "--max-retries", "1"])
+    monkeypatch.setattr(sys, "exit", lambda code=0: (_ for _ in ()).throw(SystemExit(code)))
+    # Patch wait_for_redis_ready to not raise
+    monkeypatch.setattr(redis_isready_mod, "wait_for_redis_ready", lambda **kwargs: None)
+    # Patch settings.cache_type to "redis"
+    monkeypatch.setattr(redis_isready_mod.settings, "cache_type", "redis")
+    with pytest.raises(SystemExit) as exc:
+        redis_isready_mod.main()
+    assert exc.value.code == 0
+
+
+def test_parse_cli_and_main_fail(monkeypatch):
+    """Test CLI main() with RuntimeError (exit 1)."""
+    import mcpgateway.utils.redis_isready as redis_isready_mod
+    monkeypatch.setattr(sys, "argv", ["redis_isready.py", "--max-retries", "1"])
+    monkeypatch.setattr(sys, "exit", lambda code=0: (_ for _ in ()).throw(SystemExit(code)))
+    def fail(**kwargs): raise RuntimeError("fail")
+    monkeypatch.setattr(redis_isready_mod, "wait_for_redis_ready", fail)
+    monkeypatch.setattr(redis_isready_mod.settings, "cache_type", "redis")
+    with pytest.raises(SystemExit) as exc:
+        redis_isready_mod.main()
+    assert exc.value.code == 1
+
+
+def test_main_not_using_redis(monkeypatch):
+    """If not using Redis, main() should exit 0."""
+    import mcpgateway.utils.redis_isready as redis_isready_mod
+    monkeypatch.setattr(redis_isready_mod.settings, "cache_type", "none")
+    monkeypatch.setattr(sys, "exit", lambda code=0: (_ for _ in ()).throw(SystemExit(code)))
+    # __main__ block
+    if hasattr(redis_isready_mod, "__main__"):
+        delattr(redis_isready_mod, "__main__")
+    with pytest.raises(SystemExit) as exc:
+        # Simulate __main__ block
+        if redis_isready_mod.settings.cache_type == "redis":
+            redis_isready_mod.main()
+        else:
+            sys.exit(0)
+    assert exc.value.code == 0
+
+
+def test_invalid_cli_params(monkeypatch):
+    """Test CLI with invalid params (exit 3)."""
+    import mcpgateway.utils.redis_isready as redis_isready_mod
+    monkeypatch.setattr(sys, "argv", ["redis_isready.py", "--max-retries", "0"])
+    # Patch sys.exit to catch exit code
+    monkeypatch.setattr(sys, "exit", lambda code=0: (_ for _ in ()).throw(SystemExit(code)))
+    # Patch wait_for_redis_ready to raise RuntimeError for invalid params
+    def fail(**kwargs): raise RuntimeError("Invalid max_retries or retry_interval_ms values")
+    monkeypatch.setattr(redis_isready_mod, "wait_for_redis_ready", fail)
+    monkeypatch.setattr(redis_isready_mod.settings, "cache_type", "redis")
+    with pytest.raises(SystemExit) as exc:
+        try:
+            redis_isready_mod.main()
+        except RuntimeError:
+            # If main doesn't catch, test will still pass
+            pass