如何重构Slack消息构建函数以提升其可测试性？

Great question! Your current build_list_message does way too many things at once—it handles database queries, business filtering, Slack message assembly, and external service calls (like SlackUser.retrieve). This tight coupling makes it really hard to test individual parts without hitting the database or external services. Yes, you absolutely should split this into smaller, single-responsibility functions to boost testability. Here's how to refactor it step by step:

1. Separate Data Fetching/Filtering from Message Construction #

First, extract the database query logic into its own function. This lets you test filtering rules independently without touching Slack message formatting, and vice versa.

def get_filtered_resources(team_id, resource_type='all', availability='all', channel_id=None):
    """Fetch and filter Page/AsyncTask resources based on given criteria."""
    pages = Page.objects.none()
    async_tasks = AsyncTask.objects.none()

    # Filter by resource type
    if resource_type in ['web_pages', 'all']:
        pages = Page.objects.filter(user__team__team_id=team_id).order_by('title')
    if resource_type in ['async_tasks', 'all']:
        async_tasks = AsyncTask.objects.filter(user__team__team_id=team_id).order_by('title')

    # Filter by availability
    if availability == 'available':
        pages = pages.filter(available=True)
        async_tasks = async_tasks.filter(available=True)
    elif availability == 'unavailable':
        pages = pages.filter(available=False)
        async_tasks = async_tasks.filter(available=False)

    # Filter by channel (if provided)
    if channel_id:
        pages = pages.filter(alert_channel=channel_id)
        async_tasks = async_tasks.filter(alert_channel=channel_id)

    return pages, async_tasks

2. Use Dependency Injection for External Dependencies #

Modify the main function to accept dependencies (like resource fetching or Slack user retrieval) as optional parameters. This lets you swap in mock implementations during testing, avoiding real database/Slack calls.

def build_list_message(team_id, user_id, msg_state=None, chl_state=None,
                       resource_fetcher=None, slack_user_retriever=None):
    """Orchestrate resource fetching and Slack message assembly."""
    # Set defaults for normal runtime usage
    msg_state = msg_state or {}
    chl_state = chl_state or {}
    resource_fetcher = resource_fetcher or get_filtered_resources
    slack_user_retriever = slack_user_retriever or SlackUser.retrieve

    # Extract filter parameters from state objects
    resource_type = msg_state.get('resource_type', 'all')
    availability = msg_state.get('resource_availability', 'all')
    channel_id = chl_state.get('channel_id')

    # Get filtered resources and user data
    pages, async_tasks = resource_fetcher(team_id, resource_type, availability, channel_id)
    user = slack_user_retriever(team_id, user_id)

    # Assemble Slack message attachments
    attachments = [
        _build_filters(resource_type, availability),
        *[_build_page_item(p, user) for p in pages],
        *[_build_async_task_item(at, user) for at in async_tasks]
    ]

    return {
        'text': "Here's the list of all monitoring resources",
        'attachments': attachments
    }

3. Reduce Duplication in Private Attachment Builders #

Your _build_page_item and _build_async_task_item functions are nearly identical. Extract a shared helper to cut down on repetition and make testing easier (you only need to test one core logic instead of two):

def _build_resource_item(resource, user, resource_label, callback_id):
    """Shared helper to build Slack attachments for any resource type."""
    return {
        "fallback": resource_label,
        "color": resource.status_color,
        "mrkdwn_in": ["fields"],
        "callback_id": callback_id,
        "fields": [
            {
                "title": resource.title,
                "value": f"_{resource_label}_ ({resource.status})"
            },
            {
                "title": "URL",
                "value": resource.url
            }
        ],
        "footer": _build_resource_footer(resource),
        "actions": _build_resource_item_actions(resource, user)
    }

# Update original builders to use the shared helper
def _build_page_item(page, user):
    return _build_resource_item(page, user, "Page", 'page_change')

def _build_async_task_item(async_task, user):
    return _build_resource_item(async_task, user, "Async task", 'async_task_change')

How This Improves Testability #

Now you can write focused, fast tests without external dependencies:

Example Test for Message Assembly (Using Mocks)

from unittest.mock import MagicMock

def test_build_list_message_with_filtered_pages():
    # Mock resource fetcher to return fake data
    def mock_resource_fetcher(*args):
        mock_page = MagicMock()
        mock_page.title = "Test Landing Page"
        mock_page.status_color = "#00ff00"
        mock_page.status = "Available"
        mock_page.url = "https://example.com"
        return [mock_page], []

    # Mock Slack user retrieval
    def mock_slack_user_retriever(*args):
        return MagicMock(id="U123")

    # Call the function with mocks
    result = build_list_message(
        team_id="T456",
        user_id="U789",
        msg_state={"resource_type": "web_pages", "resource_availability": "available"},
        resource_fetcher=mock_resource_fetcher,
        slack_user_retriever=mock_slack_user_retriever
    )

    # Assert the message structure is correct
    assert len(result["attachments"]) == 2
    assert result["attachments"][0]["actions"][0]["selected_options"][0]["value"] == "web_pages"
    assert result["attachments"][1]["fields"][0]["title"] == "Test Landing Page"

Example Test for Filter Logic

You can test get_filtered_resources directly to verify that filtering rules work as expected, using test databases or query mocks.

Key Takeaways #

• Single Responsibility: Each function now does one thing (fetch data, build attachments, orchestrate the workflow) making tests easier to target.
• Dependency Injection: Swapping real implementations for mocks lets you test logic in isolation, without hitting databases or external APIs.
• Reduced Duplication: Shared helpers mean less code to test and maintain.

内容的提问来源于stack exchange，提问作者ryche