建筑条形信息
建筑条形信息
提问于 2019-01-09 10:20:24
我有用text和attachments构建和返回Slack消息的函数。如何重构此函数以使其更易于测试？我应该把它分成多个函数吗？
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
}以下是私人功能：
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)
}回答 1
Code Review用户
发布于 2019-09-09 21:37:57
我们可以通过给函数显式类型签名来了解代码应该做什么。从最高层开始，我们在做什么？
team_id是某种类型的Team，它可能是一个字符串，也可能是一个int，但现在我假设它是一个TeamID。我们知道这正是我们所需要的，因为它被传递给了Django过滤器一个SlackUser.retrieve。- 同样，我们可以假设
user_id是UserID。但是，看看它是如何使用的:我们使用它所做的就是检索SlackUser。- 我们不应该调用这个函数，除非我们知道
team_id和user_id。如果我们只有一个参数user:SlackUser**?**，那么我们就不需要从这个函数中读取那些DB表了。
- 我们不应该调用这个函数，除非我们知道
msg_state应该是个白痴，但是我们不允许用户通过。这个论点可能是一个更大的问题，里面有很多其他的东西，但就目前的目的而言，它实际上只是捆绑了两个影子参数。resource_type，它本身就是web_pages:bool=True和async_tasks:bool=True的一个包。availability，同样也分为available:bool=True和unavailable:bool=True。这些不是很好的名字，我们以后可以改进它们。
chl_state只是掩盖了我们真正的参数：channel_id:Optional[ChannelID]=None。- 我们要回一个白痴，但大部分都是静态的。我们实际上正在计算的部分是一个字典列表。
- 这些家伙显然有一个其他地方需要的结构。对于如何解释和执行该结构，您有几个选项；小的继承树可能是明智的。我假设您有一个类
UIMessageItem。 - 因此，有趣的功能将返回一个
Iterable[UIMessageItem]，它将被外部的dict结构所包装。
- 这些家伙显然有一个其他地方需要的结构。对于如何解释和执行该结构，您有几个选项；小的继承树可能是明智的。我假设您有一个类
第一轮：
我将old_build_list_message作为新函数的包装器，只是为了展示如何使用新版本来代替新版本。
助手：
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')
)
}心脏：
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]
]第2轮：
- 打给
_build_filters的电话似乎不太合适。它不需要业务逻辑所需的外部状态。 - 业务逻辑似乎是一次做两件非常相似的事情。我们希望相同的部分发生一次，与不同的部分分开。
- 我们可能会将“显示”逻辑从“查找”逻辑中分离出来。
助手：
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)
)
}心脏：
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)，那会把我们留在哪里？
- 有些情景细节您必须自己决定，比如
map(_build_UI_message_item,...)是否进入了build_list_message。 django_filters()可能会变得更容易读懂。- 显然，我已经介绍了迭代工具、函数工具和打字库。他们是标准包的一部分，所以希望这没问题。
- 东西是可测试的吗？
- 如果你觉得有必要的话，
_build_filters()是纯的，很容易测试的。 _build_UI_message_item()是纯的，易于测试。- 如果仍然需要这样的方法，那么
old_build_list_message()是更深层次方法的薄包装器。给它自己的类型签名，你很可能不用为它编写单元测试就能逃脱。 django_filters()是纯的。测试并不容易，但是如果您的测试套件提供了QuerySet对象，那么您应该没事。build_list_message()是不纯的！我们把我们所有的不纯代码都整合到一起了，几乎成了一条线。为了对它进行测试，我们需要某种类型的测试数据库，或者其他一些欺骗Django管理器类的方法。这很可能是可行的。
- 如果你觉得有必要的话，
- 说到测试:我还没有测试过其中的任何一个，所以它可能不像编写的那样工作。
页面原文内容由Code Review提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://codereview.stackexchange.com/questions/211173
复制相关文章
相似问题
腾讯云开发者
