Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 12s
355 lines
7.7 KiB
Markdown
355 lines
7.7 KiB
Markdown
# Mocking
|
|
|
|
## Table of Contents
|
|
- [Basic Mocking](#basic-mocking)
|
|
- [Patching](#patching)
|
|
- [Mock Objects](#mock-objects)
|
|
- [Async Mocking](#async-mocking)
|
|
- [Fixture-Based Mocking](#fixture-based-mocking)
|
|
- [Common Patterns](#common-patterns)
|
|
|
|
## Basic Mocking
|
|
|
|
### MagicMock Basics
|
|
|
|
```python
|
|
from unittest.mock import MagicMock
|
|
|
|
def test_with_mock():
|
|
# Create a mock object
|
|
mock_service = MagicMock()
|
|
|
|
# Configure return value
|
|
mock_service.get_data.return_value = {"id": 1, "name": "Test"}
|
|
|
|
# Use mock
|
|
result = mock_service.get_data()
|
|
assert result["name"] == "Test"
|
|
|
|
# Verify call
|
|
mock_service.get_data.assert_called_once()
|
|
```
|
|
|
|
### Return Value Configuration
|
|
|
|
```python
|
|
from unittest.mock import MagicMock
|
|
|
|
mock = MagicMock()
|
|
|
|
# Simple return value
|
|
mock.method.return_value = 42
|
|
|
|
# Different returns on consecutive calls
|
|
mock.method.side_effect = [1, 2, 3]
|
|
assert mock.method() == 1
|
|
assert mock.method() == 2
|
|
assert mock.method() == 3
|
|
|
|
# Raise exception
|
|
mock.method.side_effect = ValueError("Error!")
|
|
|
|
# Dynamic return value
|
|
mock.method.side_effect = lambda x: x * 2
|
|
assert mock.method(5) == 10
|
|
```
|
|
|
|
## Patching
|
|
|
|
### patch Decorator
|
|
|
|
```python
|
|
from unittest.mock import patch
|
|
|
|
# Patch a function
|
|
@patch("mymodule.external_api")
|
|
def test_with_patched_api(mock_api):
|
|
mock_api.return_value = {"status": "ok"}
|
|
result = mymodule.call_api()
|
|
assert result["status"] == "ok"
|
|
|
|
# Patch multiple
|
|
@patch("mymodule.function_a")
|
|
@patch("mymodule.function_b")
|
|
def test_multiple_patches(mock_b, mock_a):
|
|
# Note: decorators apply bottom-up
|
|
mock_a.return_value = "a"
|
|
mock_b.return_value = "b"
|
|
```
|
|
|
|
### patch Context Manager
|
|
|
|
```python
|
|
from unittest.mock import patch
|
|
|
|
def test_with_context_manager():
|
|
with patch("mymodule.external_api") as mock_api:
|
|
mock_api.return_value = {"data": "mocked"}
|
|
result = mymodule.process()
|
|
assert result == {"data": "mocked"}
|
|
# Original function restored after with block
|
|
```
|
|
|
|
### patch.object
|
|
|
|
```python
|
|
from unittest.mock import patch
|
|
|
|
class MyService:
|
|
def fetch_data(self):
|
|
return "real data"
|
|
|
|
def test_patch_instance_method():
|
|
service = MyService()
|
|
|
|
with patch.object(service, "fetch_data", return_value="mocked"):
|
|
assert service.fetch_data() == "mocked"
|
|
```
|
|
|
|
### Patching Where Used
|
|
|
|
```python
|
|
# mymodule.py
|
|
from requests import get # 'get' is imported here
|
|
|
|
def fetch_url(url):
|
|
return get(url).text
|
|
|
|
# test_mymodule.py
|
|
# Patch where it's USED, not where it's DEFINED
|
|
@patch("mymodule.get") # NOT "requests.get"
|
|
def test_fetch_url(mock_get):
|
|
mock_get.return_value.text = "mocked response"
|
|
result = fetch_url("http://example.com")
|
|
assert result == "mocked response"
|
|
```
|
|
|
|
## Mock Objects
|
|
|
|
### Spec and Autospec
|
|
|
|
```python
|
|
from unittest.mock import MagicMock, create_autospec
|
|
|
|
class UserService:
|
|
def get_user(self, user_id: int) -> dict:
|
|
pass
|
|
|
|
def create_user(self, data: dict) -> dict:
|
|
pass
|
|
|
|
# Basic mock (allows any attribute)
|
|
mock = MagicMock()
|
|
mock.nonexistent_method() # Works, but shouldn't
|
|
|
|
# Spec mock (restricts to real attributes)
|
|
mock = MagicMock(spec=UserService)
|
|
# mock.nonexistent_method() # Raises AttributeError
|
|
|
|
# Autospec (also checks signatures)
|
|
mock = create_autospec(UserService)
|
|
# mock.get_user() # Raises TypeError (missing user_id)
|
|
mock.get_user(123) # Works
|
|
```
|
|
|
|
### Assertion Methods
|
|
|
|
```python
|
|
from unittest.mock import MagicMock, call
|
|
|
|
mock = MagicMock()
|
|
mock.method(1, 2, key="value")
|
|
mock.method(3, 4)
|
|
|
|
# Verify calls
|
|
mock.method.assert_called()
|
|
mock.method.assert_called_once() # Fails - called twice
|
|
mock.method.assert_called_with(3, 4)
|
|
mock.method.assert_any_call(1, 2, key="value")
|
|
|
|
# Check call count
|
|
assert mock.method.call_count == 2
|
|
|
|
# Check all calls
|
|
assert mock.method.call_args_list == [
|
|
call(1, 2, key="value"),
|
|
call(3, 4),
|
|
]
|
|
|
|
# Reset mock
|
|
mock.reset_mock()
|
|
assert mock.method.call_count == 0
|
|
```
|
|
|
|
## Async Mocking
|
|
|
|
### AsyncMock
|
|
|
|
```python
|
|
from unittest.mock import AsyncMock, patch
|
|
import pytest
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_mock():
|
|
mock = AsyncMock(return_value={"data": "mocked"})
|
|
result = await mock()
|
|
assert result["data"] == "mocked"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("mymodule.async_fetch", new_callable=AsyncMock)
|
|
async def test_patched_async(mock_fetch):
|
|
mock_fetch.return_value = {"status": "ok"}
|
|
result = await mymodule.process()
|
|
assert result["status"] == "ok"
|
|
```
|
|
|
|
### Async Side Effects
|
|
|
|
```python
|
|
from unittest.mock import AsyncMock
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_side_effect():
|
|
mock = AsyncMock()
|
|
|
|
# Return different values
|
|
mock.side_effect = [1, 2, 3]
|
|
assert await mock() == 1
|
|
assert await mock() == 2
|
|
|
|
# Async function as side effect
|
|
async def async_side_effect(x):
|
|
return x * 2
|
|
|
|
mock.side_effect = async_side_effect
|
|
assert await mock(5) == 10
|
|
```
|
|
|
|
## Fixture-Based Mocking
|
|
|
|
### Mock Fixtures
|
|
|
|
```python
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
@pytest.fixture
|
|
def mock_database():
|
|
"""Fixture providing mock database."""
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value = [{"id": 1}, {"id": 2}]
|
|
return mock_db
|
|
|
|
def test_with_mock_fixture(mock_database):
|
|
result = mock_database.query("SELECT * FROM users")
|
|
assert len(result) == 2
|
|
|
|
@pytest.fixture
|
|
def mock_external_api():
|
|
"""Fixture with patching."""
|
|
with patch("mymodule.external_api") as mock:
|
|
mock.return_value = {"status": "ok"}
|
|
yield mock
|
|
```
|
|
|
|
### Monkeypatch Fixture
|
|
|
|
```python
|
|
def test_with_monkeypatch(monkeypatch):
|
|
# Patch function
|
|
monkeypatch.setattr("mymodule.get_config", lambda: {"debug": True})
|
|
|
|
# Patch environment variable
|
|
monkeypatch.setenv("API_KEY", "test-key")
|
|
|
|
# Patch dictionary item
|
|
monkeypatch.setitem(mymodule.settings, "DEBUG", True)
|
|
|
|
# Patch attribute
|
|
monkeypatch.setattr(mymodule.client, "timeout", 5)
|
|
```
|
|
|
|
## Common Patterns
|
|
|
|
### Mock HTTP Responses
|
|
|
|
```python
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
@patch("requests.get")
|
|
def test_http_request(mock_get):
|
|
# Configure mock response
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"data": "test"}
|
|
mock_get.return_value = mock_response
|
|
|
|
result = fetch_data("http://api.example.com")
|
|
|
|
assert result == {"data": "test"}
|
|
mock_get.assert_called_once_with("http://api.example.com")
|
|
```
|
|
|
|
### Mock File Operations
|
|
|
|
```python
|
|
from unittest.mock import mock_open, patch
|
|
|
|
def test_read_file():
|
|
mock_file_content = "Hello, World!"
|
|
|
|
with patch("builtins.open", mock_open(read_data=mock_file_content)):
|
|
result = read_file("test.txt")
|
|
assert result == "Hello, World!"
|
|
|
|
def test_write_file():
|
|
m = mock_open()
|
|
|
|
with patch("builtins.open", m):
|
|
write_file("test.txt", "content")
|
|
|
|
m().write.assert_called_once_with("content")
|
|
```
|
|
|
|
### Mock Datetime
|
|
|
|
```python
|
|
from unittest.mock import patch
|
|
from datetime import datetime
|
|
|
|
@patch("mymodule.datetime")
|
|
def test_with_frozen_time(mock_datetime):
|
|
mock_datetime.now.return_value = datetime(2024, 1, 15, 12, 0, 0)
|
|
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
|
|
|
|
result = mymodule.get_timestamp()
|
|
assert result == "2024-01-15 12:00:00"
|
|
```
|
|
|
|
### Mock Class Instance
|
|
|
|
```python
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
class EmailService:
|
|
def send(self, to, subject, body):
|
|
# Real implementation
|
|
pass
|
|
|
|
@patch("mymodule.EmailService")
|
|
def test_email_sent(MockEmailService):
|
|
# Configure mock instance
|
|
mock_instance = MagicMock()
|
|
MockEmailService.return_value = mock_instance
|
|
|
|
# Call function that uses EmailService
|
|
send_notification("user@example.com", "Hello!")
|
|
|
|
# Verify email was sent
|
|
mock_instance.send.assert_called_once_with(
|
|
"user@example.com",
|
|
"Notification",
|
|
"Hello!",
|
|
)
|
|
```
|