"""Tests for AS400 MCP Server"""
import pytest
class TestListTables:
"""Tests for list_tables tool"""
def test_list_tables_basic(self, mock_odbc, sample_tables):
"""Test basic table listing"""
from as400_mcp.server import _list_tables_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [
("TABLE_NAME",),
("TABLE_TEXT",),
("TABLE_TYPE",),
("ROW_COUNT",),
]
mock_cursor.fetchall.return_value = sample_tables
result = _list_tables_internal("MYLIB")
assert len(result) == 3
assert result[0]["TABLE_NAME"] == "ORDER"
assert result[0]["TABLE_TEXT"] == "受注マスタ"
def test_list_tables_with_pattern(self, mock_odbc, sample_tables):
"""Test table listing with pattern filter"""
from as400_mcp.server import _list_tables_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [
("TABLE_NAME",),
("TABLE_TEXT",),
("TABLE_TYPE",),
("ROW_COUNT",),
]
# Filter to only ORDER tables
filtered = [t for t in sample_tables if t[0].startswith("ORDER")]
mock_cursor.fetchall.return_value = filtered
result = _list_tables_internal("MYLIB", pattern="ORDER%")
assert len(result) == 2
assert all("ORDER" in r["TABLE_NAME"] for r in result)
class TestListSourceFiles:
"""Tests for list_source_files tool"""
def test_list_source_files_basic(self, mock_odbc):
"""Test basic source file listing"""
from as400_mcp.server import _list_source_files_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [
("SOURCE_FILE",),
("DESCRIPTION",),
("MEMBER_COUNT",),
("CCSID",),
]
mock_cursor.fetchall.return_value = [
("QCLSRC", "CL Source", 10, 5035),
("QRPGSRC", "RPG Source", 25, 5035),
("QRPGLESRC", "RPGLE Source", 15, 5035),
]
result = _list_source_files_internal("MYLIB")
assert len(result) == 3
assert result[0]["SOURCE_FILE"] == "QCLSRC"
assert result[1]["MEMBER_COUNT"] == 25
class TestGetColumns:
"""Tests for get_columns tool"""
def test_get_columns_basic(self, mock_odbc, sample_columns):
"""Test basic column retrieval"""
from as400_mcp.server import _get_columns_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [
("COLUMN_NAME",),
("COLUMN_TEXT",),
("DATA_TYPE",),
("LENGTH",),
("DECIMAL_PLACES",),
("IS_NULLABLE",),
("ORDINAL_POSITION",),
("DEFAULT_VALUE",),
("CCSID",),
]
mock_cursor.fetchall.return_value = sample_columns
result = _get_columns_internal("MYLIB", "ORDER")
assert len(result) == 4
assert result[0]["COLUMN_NAME"] == "ORDNO"
assert result[0]["COLUMN_TEXT"] == "受注番号"
assert result[0]["DATA_TYPE"] == "DECIMAL"
def test_get_columns_japanese_labels(self, mock_odbc, sample_columns):
"""Test that Japanese labels are preserved"""
from as400_mcp.server import _get_columns_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [
("COLUMN_NAME",),
("COLUMN_TEXT",),
("DATA_TYPE",),
("LENGTH",),
("DECIMAL_PLACES",),
("IS_NULLABLE",),
("ORDINAL_POSITION",),
("DEFAULT_VALUE",),
("CCSID",),
]
mock_cursor.fetchall.return_value = sample_columns
result = _get_columns_internal("MYLIB", "ORDER")
# Verify all Japanese labels are present
labels = [r["COLUMN_TEXT"] for r in result]
assert "受注番号" in labels
assert "顧客コード" in labels
assert "受注日" in labels
class TestGetSource:
"""Tests for get_source tool"""
def test_get_source_basic(self, mock_odbc):
"""Test basic source retrieval"""
from as400_mcp.server import _get_source_internal
mock_conn, mock_cursor = mock_odbc
# First call: metadata
mock_cursor.description = [
("MEMBER_NAME",),
("SOURCE_TYPE",),
("MEMBER_TEXT",),
("LAST_UPDATED",),
]
mock_cursor.fetchone.return_value = (
"ORDMNT",
"RPGLE",
"受注メンテナンス",
"2024-01-15 10:30:00",
)
# Second call: source lines
mock_cursor.fetchall.return_value = [
(1.00, 240115, " H DFTACTGRP(*NO)"),
(2.00, 240115, " F ORDER UF E K DISK"),
(3.00, 240115, " D ORDNO S 10P 0"),
]
result = _get_source_internal("MYLIB", "QRPGSRC", "ORDMNT")
assert "metadata" in result
assert result["metadata"]["MEMBER_NAME"] == "ORDMNT"
assert result["metadata"]["SOURCE_TYPE"] == "RPGLE"
def test_get_source_not_found(self, mock_odbc):
"""Test source not found error"""
from as400_mcp.server import _get_source_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [
("MEMBER_NAME",),
("SOURCE_TYPE",),
("MEMBER_TEXT",),
("LAST_UPDATED",),
]
mock_cursor.fetchone.return_value = None
result = _get_source_internal("MYLIB", "QRPGSRC", "NOTEXIST")
assert "error" in result
class TestExecuteSql:
"""Tests for execute_sql tool"""
def test_execute_sql_select_only(self):
"""Test that only SELECT statements are allowed"""
from as400_mcp.server import _execute_sql_internal
# DELETE should be rejected
result = _execute_sql_internal("DELETE FROM MYLIB.ORDER")
assert "error" in result
assert "SELECT" in result["error"]
# UPDATE should be rejected
result = _execute_sql_internal("UPDATE MYLIB.ORDER SET ORDAMT = 0")
assert "error" in result
# INSERT should be rejected
result = _execute_sql_internal("INSERT INTO MYLIB.ORDER VALUES(1)")
assert "error" in result
def test_execute_sql_valid_select(self, mock_odbc):
"""Test valid SELECT execution"""
from as400_mcp.server import _execute_sql_internal
mock_conn, mock_cursor = mock_odbc
mock_cursor.description = [("ORDNO",), ("CUSTCD",)]
mock_cursor.fetchall.return_value = [
(1, "CUST001"),
(2, "CUST002"),
]
result = _execute_sql_internal("SELECT ORDNO, CUSTCD FROM MYLIB.ORDER")
assert "rows" in result
assert len(result["rows"]) == 2
class TestResourceURIs:
"""Tests for MCP resource URIs"""
def test_resource_uri_format(self):
"""Test that resource URIs follow expected format"""
# Resource URIs should follow pattern:
# as400://library/{lib}/tables
# as400://library/{lib}/table/{tbl}/schema
# as400://library/{lib}/sources
# as400://library/{lib}/source/{srcf}/{mbr}
expected_patterns = [
"as400://library/MYLIB/tables",
"as400://library/MYLIB/table/ORDER/schema",
"as400://library/MYLIB/sources",
"as400://library/MYLIB/source/QRPGSRC/ORDMNT",
]
for pattern in expected_patterns:
assert pattern.startswith("as400://")
assert "/library/" in pattern
# Mark tests that require actual ODBC connection
@pytest.mark.requires_odbc
class TestIntegration:
"""Integration tests - require actual AS400 connection"""
def test_real_connection(self):
"""Test with real AS400 connection"""
pytest.skip("Requires AS400_CONNECTION_STRING environment variable")
