Building Slack message

Question

I have the function which builds and return Slack message with text and attachments. How can I refactor this function to make it easier to test? Should I split it into multiple functions?

def build_list_message(team_id, user_id, msg_state=None, chl_state=None):
    if not msg_state:
        msg_state = {}

    if not chl_state:
        chl_state = {}

    resource_type = msg_state.get('resource_type', 'all')
    availability = msg_state.get('resource_availability', 'all')

    pages = Page.objects.none()
    async_tasks = AsyncTask.objects.none()

    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')

    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)

    channel_id = chl_state.get('channel_id')
    if channel_id:
        pages = pages.filter(alert_channel=channel_id)
        async_tasks = async_tasks.filter(alert_channel=channel_id)

    user = SlackUser.retrieve(team_id, user_id)

    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
    }

Here is private functions:

def _build_filters(resource_type, availability):
    resource_types = [
        {"text": "All types", "value": "all"},
        {"text": ":link: Webpages", "value": "web_pages"}
    ]

    availability_choices = [
        {"text": "Available / Unavailable", "value": "all"},
        {"text": ":white_circle: Available", "value": "available"},
        {"text": ":red_circle: Unavaliable", "value": "unavailable"}
    ]

    selected_resource_types = list(filter(
        lambda t: t['value'] == resource_type, resource_types))

    selected_availability_choices = list(filter(
        lambda a: a['value'] == availability, availability_choices))

    return {
        "fallback": "Resource filters",
        "color": "#d2dde1",
        "mrkdwn_in": ["text"],
        "callback_id": "resource_filters",
        "actions": [
            {
                "name": "resource_type",
                "text": "Type",
                "type": "select",
                "options": resource_types,
                "selected_options": selected_resource_types
            },
            {
                "name": "resource_availability",
                "text": "Available",
                "type": "select",
                "options": availability_choices,
                "selected_options": selected_availability_choices
            }
        ]
    }


def _build_page_item(page, user):
    return {
        "fallback": "Page",
        "color": page.status_color,
        "mrkdwn_in": ["fields"],
        "callback_id": 'page_change',
        "fields": [
            {
                "title": page.title,
                "value": f"_Page_ ({page.status})"
            },
            {
                "title": "URL",
                "value": page.url
            }
        ],
        "footer": _build_resource_footer(page),
        "actions": _build_resource_item_actions(page, user)
    }


def _build_async_task_item(async_task, user):
    return {
        "fallback": "Async task",
        "color": async_task.status_color,
        "mrkdwn_in": ["fields"],
        "callback_id": 'async_task_change',
        "fields": [
            {
                "title": async_task.title,
                "value": f"_Async task_ ({async_task.status})"
            },
            {
                "title": "URL",
                "value": async_task.url
            }
        ],
        "footer": _build_resource_footer(async_task),
        "actions": _build_resource_item_actions(async_task, user)
    }

Answer 1

I've come back to this question about 4 or 5 times since it was originally posted, started writing up an answer, started doubting my alternative implementation, scrapped it, and then come back about a month later. This tells me a few things:

1. You have a tricky problem, where the "best" solution is not necessarily intuitive, and how you measure "best" can vary wildly
2. Overall, the code is pretty good, so I can't just write a more nit-picky answer and ignore the larger question of "is this structured the right way"

Because I had such a hard time with this, I'm not going to completely rewrite this, or provide an alternate implementation, or anything like that. Instead, I'm going to call out some overall architecture and design choices that I think have made this tricky, some potential solutions with pros/cons, and then leave it as an exercise for the reader to decide what they want to do.

I'll also include a few nitpicky things about the code, because that's just who I am as a person and I'll feel bad if I don't review your code at all.

Lastly, I've never used Django before, so its very possible a much better solution exists within the framework, or that someone else has solved the problem in another 3rd party library, or whatever. I encourage you to do your own research on if Django can make this easier.

---

The problem

How can I refactor this function to make it easier to test?

You're right that this is hard to test as-is. The most significant reason for this, IMO, is that you're mixing up a lot of magic, a lot of state that your function doesn't own, and the business logic that your function does own.

Magic

Your function(s) are littered with magic numbers, strings, etc. These are all things that apparently have significant meaning, and are very tightly coupled with your implementation. If you ever decide that you want your colors to change, or you want to change somethings title, or the default resource type, or whatever, then you have a lot of logic that assumes the magic.

A good way to make this less magic, easier to test, and easier to update is to at least make all magic values constants at the beginning of the file. For example, if you do this msg_state.get('resource_type', 'all') then it seems a little unclear what we're doing; just some place-holder, or something more meaningful? But if you do this msg_state.get('resource_type', DEFAULT_RESOURCE_TYPE) then it becomes immediately clear.

Note that you wouldn't want to do this msg_state.get('resource_type', ALL_RESOURCE_TYPE) because that isn't really any better than just putting 'all' there, unless you think that the resource type meaning all would change values.

This is also a good time to propose using enums instead of strings, as you get runtime validation of correct values (e.g. ResourceTypeEnum.lAl will give you an error that lAl is not a valid enum, while 'lAl' won't warn you that you typoed your string.

State

Your function to build a message depends a lot on the state of your application, of third party libraries, the framework, etc. In just this one function, you explicitly query the following kinds of state:

1. Django state (pages and tasks)
2. Slack state (users)
3. Application state (message, channels, teams)

This couples you really tightly with these things that your function can't control, and is the death of any truly repeatable unit testing. There are a lot of ways you could address them; for example you could make a service per state and use dependency injection or some other inversion of control methodology. This is what I was playing around with for a while, and eventually came up with a 500 line, OOPomination that was super generic and super unreadable. I don't know what support Django has for this, but I would guess that at some point someone has made dependency injection easier.

Ultimately, I think there is room for some level of encapsulation here. It is likely worthwhile to create separate methods for the different types of work, and filtering, etc. Could you subclass Page or AsyncTask to provide a slack_item property? Is there a better spot in your application to filter on resource type, or team, or availability, or channel, etc?

Business Logic

Once you peel everything back, your actual logic is pretty straightforward:

1. Get the list of resources that meet some filtering criteria
2. Transform them into json that meets some Slack specification
3. Send it back

This part is pretty straightforward, and I don't have a lot to say about it. These each sound like good candidates for unit testing though, and might be worth splitting up that way.

A potential solution

I think whatever you do to restructure this is going to add more code and complexity (by becoming more generic/loosely coupled), and how much of that makes sense for your use-case depends. If your tool is meant to just be a quick-and-dirty tool for your team, it might not make sense to split up the code. If you want this to be around longer, or there are plans for future enhancements, this might be a good time to break things up. I think I'd roughly recommend the following:

1. For different kinds of data (e.g. Page vs AsyncTask vs other stuff), create different methods/handlers/classes/whatever for each one that is responsible for transforming it into the Slack format json. A perfect solution would be able to take advantage of duck-typing, where you could just do input_object.to_slack_json()
2. Play around with the filtering - can this be done by the caller instead of your application? Can you encapsulate how to do it in the same per-type method/handler/class/whatever as previously suggested?
3. Instead of getting your state within the function, pass it in instead. Then it can be truly independent of how it is being run, and the unit test can construct the input as desired.

---

As promised, here are some

Little nitpicky things

When doing this:

if not msg_state:
    msg_state = {}

It may be easier to write it this way: msg_state = msg_state or {}. or will pick the first truthy-element, or the last non-truthy element (e.g. the empty dictionary).

---

Instead of using this (and needlessly building lists just to throw them away)

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]
]

Consider using itertools.chain:

attachments = list(
    itertools.chain(
        (_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)
    )
)

---

Instead of using list + filter, and lambda functions that add just a bit of extra mental overhead

selected_resource_types = list(filter(
    lambda t: t['value'] == resource_type, resource_types))

selected_availability_choices = list(filter(
    lambda a: a['value'] == availability, availability_choices))

Use a list comprehension

selected_resource_types = [
    rt for rt in resource_types if resource_type["value"] == rt
]

selected_availability_choices = [
    ac for ac in availability_choices if availability["value"] == ac 
]

Or, define a class that represents a Slack option, or a selection of Slack options, and encapsulate this logic in that class. Then you don't have to repeat yourself between resource_types and availability_choices.

Rough outline:

class SlackOption:

    def __init__(self, text, value):
        self.text = text
        self.value = value

class SlackOptions:

    def __init__(self, choices):
        self.choices = choices

    def get_selection(self, choice):
        return [option for option in self.choices if option.value == choice]

Answer 2

We can learn a lot about what code is supposed to do by giving our functions explicit type signatures. Starting at the top, what are we working with?

• team_id is some kind of Team ID. It's probably a string, or maybe an int, but for now I'll assume it's a TeamID. We know it's exactly what we need because it's passed as-is to the Django filters an SlackUser.retrieve.
• user_id we can similarly assume is a UserID. However, look at how it's used: All we do with it is retrieve the SlackUser.
  • We shouldn't be calling this function unless we know the team_id and the user_id. What if we just had one argument user:SlackUser? Then we wouldn't need to read those DB tables from inside this function.
• msg_state is supposed to be a dict, but we're letting the user pass in None. Possibly this argument might be a bigger dict with lots of other stuff in it, but for the purpose at hand it really just bundles up two shadow arguments.
  • resource_type, which itself is just a bundle of web_pages:bool=True and async_tasks:bool=True.
  • availability, which likewise breaks out into available:bool=True and unavailable:bool=True. Those aren't great names; we can improve on them later.
• chl_state is just masking our real parameter: channel_id:Optional[ChannelID]=None.
• We're returning a dict, but most of it's static. The part that we're actually computing is a list of dicts.
  • Those dicts clearly have a structure that's needed elsewhere. You have a few options for how to explain and enforce that structure; a small inheritance tree is probably advisable. I'll just assume you've got a class UIMessageItem.
  • So the interesting functionality will return an Iterable[UIMessageItem], and that will get wrapped by the outer dict structure you've got.

Round 1:

I've included old_build_list_message, as a wrapper to the new function, just to show how you would use the new version in place of the new one.

helpers:

def _build_filters(*,
                   include_web_pages: bool = True,
                   include_async_tasks: bool = True,
                   include_available: bool = True,
                   include_unavailable: bool = True)
                   -> UIMessageItem:
    ...


def _build_page_item(page: Page, user: SlackUser) -> UIMessageItem:
    ...


def _build_async_task_item(task: AsyncTask, user: SlackUser) -> UIMessageItem:
    ...

def old_build_list_message(team_id, user_id, msg_state, chl_state):
    return {
        'text': "Here's the list of all monitoring resources",
        'attachments': build_list_message(
            SlackUser.retrieve(team_id, user_id),
            include_web_pages: msg_state['resource_type'] in ['web_pages', 'all'],
            include_async_tasks: msg_state['resource_type'] in ['async_tasks', 'all'],
            include_available: msg_state['resource_availability'] in ['available', 'all'],
            include_unavailable: msg_state['resource_availability'] in ['unavailable', 'all'],
            channel_id: chl_state.get('channel_id')
        )
    }

heart:

def build_list_message(user: SlackUser,
                       *,
                       include_web_pages: bool = True,
                       include_async_tasks: bool = True,
                       include_available: bool = True,
                       include_unavailable: bool = True,
                       channel_id: Optional[ChannelID] = None)
                       -> Iterable[UIMessageItem]:

    pages = Page.objects.none()
    async_tasks = AsyncTask.objects.none()

    if include_web_pages:
        pages = Page.objects.filter(
            user__team__team_id=user.team_id).order_by('title')

    if include_async_tasks:
        async_tasks = AsyncTask.objects.filter(
            user__team__team_id=user.team_id).order_by('title')

    if not include_unavailable:
        pages = pages.filter(available=True)
        async_tasks = async_tasks.filter(available=True)

    if not include_available:
        pages = pages.filter(available=False)
        async_tasks = async_tasks.filter(available=False)

    if channel_id:
        pages = pages.filter(alert_channel=channel_id)
        async_tasks = async_tasks.filter(alert_channel=channel_id)

    return [
        _build_filters(include_web_pages = include_web_pages,
                       include_async_tasks = include_async_tasks,
                       include_available = include_available,
                       include_unavailable = include_unavailable),
        *[_build_page_item(p, user) for p in pages],
        *[_build_async_task_item(at, user) for at in async_tasks]
    ]

Round 2:

• The call to _build_filters seems out of place. It doesn't need the external state that our business logic needs.
• The business logic seems to be doing two very similar things at once. We'd like the parts that are the same to happen once, separately from the parts that are different.
• We could probably break the "display" logic out from the "find" logic.

helpers:

def _build_filters(whatever):
    '''At this point this is the problem of whoever's higher in the stack.'''
    ...


def _build_UI_message_item(item: Union[Page, AsyncTask],
                           user: SlackUser)
                           -> UIMessageItem:
    ...

def old_build_list_message(team_id, user_id, msg_state, chl_state):
    items = build_list_message(
        SlackUser.retrieve(team_id, user_id),
        django_filter = django_filters(
            include_available = msg_state['resource_availability'] in ['available', 'all'],
            include_unavailable = msg_state['resource_availability'] in ['unavailable', 'all'],
            channel_id = chl_state.get('channel_id')
        ),
        include_web_pages = msg_state['resource_type'] in ['web_pages', 'all'],
        include_async_tasks = msg_state['resource_type'] in ['async_tasks', 'all']
    )
    filters_message = _build_filters(msg_state)

    return {
        'text': "Here's the list of all monitoring resources",
        'attachments': itertools.chain((filters_message, ),
                                       map(_build_UI_message_item, items)
                       )
    }

heart:

def django_filters(include_available: bool = True,
                   include_unavailable: bool = True,
                   channel_id: Optional[ChannelID] = None)
                   -> Callable[[QuerySet], QuerySet]:
    filters = {}

    if not include_unavailable:
        filters['available' = True)

    if not include_available:
        filters['available' = False)

    if channel_id:
        filters['alert_channel' = channel_id)

    return functools.partial(QuerySet.filter, **filters)

def build_list_message(user: SlackUser,
                        *,
                        django_filter: Callable[[QuerySet], QuerySet] = lambda qs: qs,
                        include_web_pages: bool = True,
                        include_async_tasks: bool = True)
                        -> Iterable[Union[Page, AsyncTask]]:
    pages = django_filter(
        Page.objects.all() if include_web_pages else Page.objects.none()
    )
    tasks = django_filter(
        AsyncTask.objects.all() if include_async_tasks else AsyncTask.objects.none()
    )

    return itertools.chain(pages, async_tasks)

Where does that leave us?

• There are some situational details you'll have to decide for yourself, like whether map(_build_UI_message_item,...) goes inside build_list_message.
• django_filters() could maybe be made easier to read somehow.
• Obviously I've introduced the itertools, functools, and typing libraries. They're part of the standard package, so hopefully that's ok.
• Is stuff testable?
  • _build_filters() is pure and easy to test, if you feel it's necessary.
  • _build_UI_message_item() is pure and easy to test.
  • old_build_list_message(), if any such method is still needed, is a thin wrapper for the deeper methods. Give it its own type-signature, and you can probably get away without writing unit tests for it.
  • django_filters() is pure. It's not super easy to test, but if your test suite has a supply of QuerySet objects then you should be ok.
  • build_list_message() is impure! We've consolidated all our impure code down, almost to a one-liner. In order to have tests for it we'll need to have some kind of test database, or some other way of spoofing the Django Manager classes. That's probably do-able.
• Speaking of test: I haven't tested any of this, so it probably doesn't work as written.