jpmschweitzerandClaude Opus 4.5
64574bcc39
Build and Push / build (release) Failing after 17s
Extracted standalone scheduler service with: - FastAPI REST API for task management - APScheduler-based task execution - PostgreSQL persistence - Docker container support - Gitea Actions CI/CD workflow 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
632 lines
21 KiB
Python
632 lines
21 KiB
Python
"""
|
|
Comprehensive tests for the generic REST API executor.
|
|
|
|
Tests cover:
|
|
- All HTTP methods (GET, POST, PUT, DELETE, PATCH)
|
|
- Authentication types (Bearer, Basic, API Key)
|
|
- Environment variable substitution
|
|
- Error handling
|
|
- Response extraction
|
|
- Sensitive data redaction
|
|
|
|
Run with: pytest tests/test_rest_api_executor.py -v -s
|
|
"""
|
|
import pytest
|
|
import asyncio
|
|
import os
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
import httpx
|
|
|
|
from src.executors import rest_api_executor
|
|
from src.config import Settings
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestRestApiExecutor:
|
|
"""Tests for rest_api_executor module."""
|
|
|
|
@pytest.fixture
|
|
def mock_httpx_client(self):
|
|
"""Mock httpx AsyncClient."""
|
|
return AsyncMock()
|
|
|
|
@pytest.fixture
|
|
def base_config(self):
|
|
"""Base configuration for REST API calls."""
|
|
return {
|
|
"url": "http://test-service:8080/api/endpoint",
|
|
"method": "POST"
|
|
}
|
|
|
|
# Basic HTTP Method Tests
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_request_success(self, test_settings: Settings):
|
|
"""Test successful POST request."""
|
|
config = {
|
|
"url": "http://httpbin.org/post",
|
|
"method": "POST",
|
|
"payload": {"test": "data"},
|
|
"timeout": 10
|
|
}
|
|
|
|
# Mock httpx response
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Success"}
|
|
mock_response.text = "Success"
|
|
|
|
mock_client.return_value.__aenter__.return_value.post = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert "Success" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_request_success(self, test_settings: Settings):
|
|
"""Test successful GET request."""
|
|
config = {
|
|
"url": "http://httpbin.org/get",
|
|
"method": "GET",
|
|
"timeout": 10
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"result": "GET success"}
|
|
mock_response.text = "Success"
|
|
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert result is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_put_request_success(self, test_settings: Settings):
|
|
"""Test successful PUT request."""
|
|
config = {
|
|
"url": "http://httpbin.org/put",
|
|
"method": "PUT",
|
|
"payload": {"update": "data"}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Updated"}
|
|
mock_response.text = "Updated"
|
|
|
|
mock_client.return_value.__aenter__.return_value.put = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert "Updated" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_request_success(self, test_settings: Settings):
|
|
"""Test successful DELETE request."""
|
|
config = {
|
|
"url": "http://httpbin.org/delete",
|
|
"method": "DELETE"
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 204
|
|
mock_response.json.side_effect = Exception("No JSON")
|
|
mock_response.text = ""
|
|
|
|
mock_client.return_value.__aenter__.return_value.delete = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert "204" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_request_success(self, test_settings: Settings):
|
|
"""Test successful PATCH request."""
|
|
config = {
|
|
"url": "http://httpbin.org/patch",
|
|
"method": "PATCH",
|
|
"payload": {"field": "value"}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Patched"}
|
|
mock_response.text = "Patched"
|
|
|
|
mock_client.return_value.__aenter__.return_value.patch = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert "Patched" in result
|
|
|
|
# Authentication Tests
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bearer_auth(self, test_settings: Settings):
|
|
"""Test Bearer token authentication."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"auth": {
|
|
"type": "bearer",
|
|
"token": "test-token-123"
|
|
}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Authenticated"}
|
|
mock_response.text = "Authenticated"
|
|
|
|
mock_get = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.get = mock_get
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify Authorization header was set
|
|
call_args = mock_get.call_args
|
|
headers = call_args.kwargs.get('headers', {})
|
|
assert 'Authorization' in headers
|
|
assert headers['Authorization'] == 'Bearer test-token-123'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basic_auth(self, test_settings: Settings):
|
|
"""Test Basic authentication."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"auth": {
|
|
"type": "basic",
|
|
"username": "testuser",
|
|
"password": "testpass"
|
|
}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Authenticated"}
|
|
mock_response.text = "Authenticated"
|
|
|
|
mock_get = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.get = mock_get
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify Authorization header was set
|
|
call_args = mock_get.call_args
|
|
headers = call_args.kwargs.get('headers', {})
|
|
assert 'Authorization' in headers
|
|
assert headers['Authorization'].startswith('Basic ')
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_key_auth(self, test_settings: Settings):
|
|
"""Test API Key authentication."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"auth": {
|
|
"type": "api_key",
|
|
"key": "my-api-key-123",
|
|
"header": "X-API-Key"
|
|
}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Authenticated"}
|
|
mock_response.text = "Authenticated"
|
|
|
|
mock_get = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.get = mock_get
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify API key header was set
|
|
call_args = mock_get.call_args
|
|
headers = call_args.kwargs.get('headers', {})
|
|
assert 'X-API-Key' in headers
|
|
assert headers['X-API-Key'] == 'my-api-key-123'
|
|
|
|
# Environment Variable Substitution Tests
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_env_var_substitution_in_url(self, test_settings: Settings):
|
|
"""Test environment variable substitution in URL."""
|
|
os.environ['TEST_SERVICE_URL'] = 'http://my-service:8080'
|
|
|
|
config = {
|
|
"url": "${TEST_SERVICE_URL}/api/endpoint",
|
|
"method": "GET"
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"message": "Success"}
|
|
mock_response.text = "Success"
|
|
|
|
mock_get = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.get = mock_get
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify URL was substituted
|
|
call_args = mock_get.call_args
|
|
url = call_args.args[0] if call_args.args else None
|
|
assert url == 'http://my-service:8080/api/endpoint'
|
|
|
|
del os.environ['TEST_SERVICE_URL']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_env_var_substitution_in_auth_token(self, test_settings: Settings):
|
|
"""Test environment variable substitution in auth token."""
|
|
os.environ['API_TOKEN'] = 'secret-token-from-env'
|
|
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"auth": {
|
|
"type": "bearer",
|
|
"token": "${API_TOKEN}"
|
|
}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {}
|
|
mock_response.text = "Success"
|
|
|
|
mock_get = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.get = mock_get
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify token was substituted
|
|
call_args = mock_get.call_args
|
|
headers = call_args.kwargs.get('headers', {})
|
|
assert headers['Authorization'] == 'Bearer secret-token-from-env'
|
|
|
|
del os.environ['API_TOKEN']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_env_var_substitution_in_payload(self, test_settings: Settings):
|
|
"""Test environment variable substitution in payload."""
|
|
os.environ['DATABASE_NAME'] = 'test_db'
|
|
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "POST",
|
|
"payload": {
|
|
"database": "${DATABASE_NAME}",
|
|
"action": "backup"
|
|
}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {}
|
|
mock_response.text = "Success"
|
|
|
|
mock_post = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.post = mock_post
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify payload was substituted
|
|
call_args = mock_post.call_args
|
|
payload = call_args.kwargs.get('json', {})
|
|
assert payload['database'] == 'test_db'
|
|
|
|
del os.environ['DATABASE_NAME']
|
|
|
|
# Response Extraction Tests
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_response_path_extraction(self, test_settings: Settings):
|
|
"""Test extracting specific field from response."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"response_path": "result.message"
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"result": {
|
|
"message": "Extracted message",
|
|
"other": "ignored"
|
|
}
|
|
}
|
|
mock_response.text = "Success"
|
|
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert result == "Extracted message"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_response_path_not_found(self, test_settings: Settings):
|
|
"""Test response path extraction when field doesn't exist."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"response_path": "result.missing"
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"result": {"other": "data"}}
|
|
mock_response.text = "Success"
|
|
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Should fallback to default message
|
|
assert result is not None
|
|
|
|
# Error Handling Tests
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_url_raises_error(self, test_settings: Settings):
|
|
"""Test that missing URL raises ValueError."""
|
|
config = {
|
|
"method": "GET"
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Missing required config: 'url'"):
|
|
await rest_api_executor.execute(config, test_settings)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_http_method_raises_error(self, test_settings: Settings):
|
|
"""Test that invalid HTTP method raises ValueError."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "INVALID"
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Invalid HTTP method"):
|
|
await rest_api_executor.execute(config, test_settings)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_error_status_code(self, test_settings: Settings):
|
|
"""Test handling of HTTP error status codes."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET"
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 500
|
|
mock_response.text = "Internal Server Error"
|
|
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
with pytest.raises(Exception, match="API call failed with status 500"):
|
|
await rest_api_executor.execute(config, test_settings)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_success_codes(self, test_settings: Settings):
|
|
"""Test custom success codes configuration."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "POST",
|
|
"success_codes": [201, 202]
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 201
|
|
mock_response.json.return_value = {"message": "Created"}
|
|
mock_response.text = "Created"
|
|
|
|
mock_client.return_value.__aenter__.return_value.post = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
assert "Created" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_error_handling(self, test_settings: Settings):
|
|
"""Test handling of network errors."""
|
|
config = {
|
|
"url": "http://unreachable-service/api",
|
|
"method": "GET",
|
|
"timeout": 1
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
side_effect=httpx.ConnectError("Connection refused")
|
|
)
|
|
|
|
with pytest.raises(Exception, match="Request failed"):
|
|
await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Sensitive Data Redaction Tests
|
|
|
|
def test_redact_sensitive_data_in_dict(self):
|
|
"""Test redaction of sensitive fields in dictionaries."""
|
|
data = {
|
|
"username": "user",
|
|
"password": "secret123",
|
|
"api_key": "key123",
|
|
"normal_field": "visible"
|
|
}
|
|
|
|
redacted = rest_api_executor._redact_sensitive(data)
|
|
|
|
assert redacted["username"] == "user"
|
|
assert redacted["password"] == "***REDACTED***"
|
|
assert redacted["api_key"] == "***REDACTED***"
|
|
assert redacted["normal_field"] == "visible"
|
|
|
|
def test_redact_sensitive_nested(self):
|
|
"""Test redaction in nested structures."""
|
|
data = {
|
|
"config": {
|
|
"database": "mydb",
|
|
"password": "secret",
|
|
"auth": {
|
|
"token": "bearer123"
|
|
}
|
|
}
|
|
}
|
|
|
|
redacted = rest_api_executor._redact_sensitive(data)
|
|
|
|
assert redacted["config"]["database"] == "mydb"
|
|
assert redacted["config"]["password"] == "***REDACTED***"
|
|
assert redacted["config"]["auth"]["token"] == "***REDACTED***"
|
|
|
|
# Helper Function Tests
|
|
|
|
def test_extract_json_path_simple(self):
|
|
"""Test simple JSONPath extraction."""
|
|
data = {"message": "test"}
|
|
result = rest_api_executor._extract_json_path(data, "message")
|
|
assert result == "test"
|
|
|
|
def test_extract_json_path_nested(self):
|
|
"""Test nested JSONPath extraction."""
|
|
data = {"result": {"status": {"message": "success"}}}
|
|
result = rest_api_executor._extract_json_path(data, "result.status.message")
|
|
assert result == "success"
|
|
|
|
def test_extract_json_path_not_found(self):
|
|
"""Test JSONPath extraction when path doesn't exist."""
|
|
data = {"message": "test"}
|
|
result = rest_api_executor._extract_json_path(data, "missing.path")
|
|
assert result is None
|
|
|
|
def test_substitute_env_vars(self):
|
|
"""Test environment variable substitution."""
|
|
os.environ['TEST_VAR'] = 'test_value'
|
|
|
|
result = rest_api_executor._substitute_env_vars("Prefix ${TEST_VAR} suffix")
|
|
assert result == "Prefix test_value suffix"
|
|
|
|
del os.environ['TEST_VAR']
|
|
|
|
def test_substitute_env_vars_missing(self):
|
|
"""Test substitution with missing environment variable."""
|
|
result = rest_api_executor._substitute_env_vars("Prefix ${MISSING_VAR} suffix")
|
|
# Should replace with empty string
|
|
assert result == "Prefix suffix"
|
|
|
|
def test_substitute_env_vars_recursive(self):
|
|
"""Test recursive environment variable substitution."""
|
|
os.environ['HOST'] = 'localhost'
|
|
os.environ['PORT'] = '8080'
|
|
|
|
data = {
|
|
"url": "http://${HOST}:${PORT}/api",
|
|
"nested": {
|
|
"key": "${HOST}"
|
|
}
|
|
}
|
|
|
|
result = rest_api_executor._substitute_env_vars_recursive(data)
|
|
|
|
assert result["url"] == "http://localhost:8080/api"
|
|
assert result["nested"]["key"] == "localhost"
|
|
|
|
del os.environ['HOST']
|
|
del os.environ['PORT']
|
|
|
|
# Configuration Options Tests
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_headers(self, test_settings: Settings):
|
|
"""Test adding custom headers."""
|
|
config = {
|
|
"url": "http://test-service/api",
|
|
"method": "GET",
|
|
"headers": {
|
|
"X-Custom-Header": "custom-value",
|
|
"User-Agent": "Test-Agent"
|
|
}
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {}
|
|
mock_response.text = "Success"
|
|
|
|
mock_get = AsyncMock(return_value=mock_response)
|
|
mock_client.return_value.__aenter__.return_value.get = mock_get
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify custom headers were set
|
|
call_args = mock_get.call_args
|
|
headers = call_args.kwargs.get('headers', {})
|
|
assert headers['X-Custom-Header'] == 'custom-value'
|
|
assert headers['User-Agent'] == 'Test-Agent'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssl_verification_disabled(self, test_settings: Settings):
|
|
"""Test disabling SSL verification."""
|
|
config = {
|
|
"url": "https://test-service/api",
|
|
"method": "GET",
|
|
"verify_ssl": False
|
|
}
|
|
|
|
with patch('httpx.AsyncClient') as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {}
|
|
mock_response.text = "Success"
|
|
|
|
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
|
return_value=mock_response
|
|
)
|
|
|
|
result = await rest_api_executor.execute(config, test_settings)
|
|
|
|
# Verify SSL verification was disabled
|
|
client_call = mock_client.call_args
|
|
assert client_call.kwargs.get('verify') is False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "-s"])
|