Na minha visão, uma automação que não consegue se comunicar efetivamente com seus stackholders tende a falhar em aspectos cruciais, como visibilidade, priorização e relevância. Embora a importância da automação seja clara para a equipe técnica, surge a questão: e para aqueles que não são especialistas no assunto?
Uma comunicação eficaz é essencial para que a automação seja valorizada por todos os envolvidos no domínio, gerando percepções e benefícios distintos. Enquanto para a equipe técnica, o principal valor reside na garantia da qualidade contínua e na prevenção de bugs críticos, para gerentes, diretores e até mesmo líderes executivos, é essencial ter acesso a informações consolidadas e concisas.
Portanto, ao concluir os testes, é importante adaptar a comunicação conforme o público-alvo. Para a equipe técnica, é válido apresentar logs de execução detalhados, incluindo passo a passo dos cenários, motivos de erros, requisições executadas e capturas de tela. Por outro lado, para as camadas superiores, a ênfase deve estar em informações sumarizadas e relevantes.
Pensando nisso, decidi criar um Listener personalizado para o Robot Framework, utilizando a API do Slack, a fim de enviar notificações específicas para os testes executados nas pipelines dos projetos em que trabalho.
Embora a famosa biblioteca RobotNotifications, eu senti a necessidade de uma abordagem diferente. Então decidi desenvolver meu próprio Listener, para atender as minhas necessidades.
OBS 1: Não vou abordar o funcionamento de forma detalhada da Slack API, pois sairia demais do contexto desse post, tudo bem ✌?
OBS 2: O código demonstrado nesse post se adapta a minha realidade. Assim como no post “Criei uma Biblioteca de Requisições e Fluxos de Testes” a ideia é mostrar mais uma coisa legal que você pode conseguir fazer com o Robot Framework, e não um simples copia e cola aí … beleza ✌?
O que é um Listener?
Antes de demonstrar o código e o funcionamento dele, vamos primeiro entender o que é um listener.
Um Listener é uma interface que permite capturar eventos e informações durante a execução dos testes. Ele atua como um observador dos testes, permitindo a implementação de lógicas adicionais para manipular e processar esses eventos.
Os Listeners podem ser usados para diversas finalidades, como gerar relatórios customizados,
enviar notificações em tempo real, integrar com outras ferramentas ou sistemas, coletar métricas, registrar logs detalhados, etc.
Durante a criação de um Listener, é possível definir quais eventos são monitorados e como a lógica de manipulação desses eventos é implementada. Isso oferece flexibilidade para adaptar as saídas de uma execução de teste às necessidades específicas de um projeto, permitindo a personalização e aprimoramento da comunicação, geração de insights e tomada de decisões com base nos resultados dos testes, dentre outros.
Slack App (Webhook)
Para que tudo isso funcione perfeitamente, precisaremos criar um app de webhook no Slack e associá-lo à um canal da sua escolha. Para que esse post não fique longo demais e não saia do objetivo, caso você não saiba como criar um webhook no slack, clique aqui, para ver um tutorial bem legal do pessoal do Slack. Caso tenha dificuldade, chame o administrador do Slack da sua empresa.
Vamos ao que interessa…
O meu projeto foi inicializado utilizando o Poetry, e possui as seguintes dependências:
- robotframework = “^5.0.1“
- slack-sdk = “3.21.2“
Na empresa em que trabalho no momento em que escrevo esse post, existem automações criadas como o Robot Framework 5.0.1 e outros já com o Robot Framework 6. Então para não obrigar uma atualização de versão do Robot Framework somente por causa da minha biblioteca, deixei explicito no pyptojecy.toml a versão mínima aceitável do Robot Framework.
No caso do Slack, decidi travar na última versão na data que desenvolvi a biblioteca, para evitar problemas futuros de incompatibilidade.
Com tudo instalado, o código ficou dessa forma:
- __init__.py
import slack_sdk
from RobotSlackNotification.const_messages import principal_block, start_suite_message, tread_error_message
class RobotSlackNotification:
ROBOT_LISTENER_API_VERSION = 3
ROBOT_LIBRARY_SCOPE = 'GLOBAL'
ROBOT_LIBRARY_VERSION = '1.2.0'
def __init__(self,
slack_token: str,
channel_id: str,
application: str,
environment: str,
branch: str,
cicd_url: str,
cicd_id: str,
devicefarm_url: str,
frontend_test = False):
self.slack_token = slack_token
self.channel_id = channel_id
self.application = application
self.environment = environment
self.branch = branch
self.cicd_url = cicd_url
self.cicd_id = cicd_id
self.devicefarm_url = devicefarm_url
self.frontend_test = frontend_test
self.client = slack_sdk.WebClient(token=slack_token)
self.ROBOT_LIBRARY_LISTENER = self
self.message_timestamp = []
self.text_fallback = f'Aplicación en prueba: {self.application}'
def start_suite(self, data, result):
message = self._build_principal_message(result, self.application, self.environment, 0, 0, 0, 0)
ts = self._post_principal_message(result, message)
self.message_timestamp.append(ts)
def end_suite(self, data, result):
statistics = self._robot_statistic(result.statistics)
count_pass = statistics.passed
count_failed = statistics.failed
count_skipped = statistics.skipped
count_total = statistics.total
message = self._build_principal_message(result, self.application, self.environment, count_total, count_pass, count_failed,
count_skipped)
self._update_principal_message(result, self.message_timestamp[0], message)
def end_test(self, data, result):
attachment_color = self._attachment_color(result)
message = self._build_thread_message(result, attachment_color)
self._post_thread_message(result, message, self.message_timestamp[0])
def _robot_statistic(self, statistics):
try:
if statistics.total:
return statistics # robotframework < 4.0.0
except:
return statistics.all # robotframework > 4.0.0
def _post_principal_message(self, result, message: str):
if not result.parent:
response = self.client.chat_postMessage(channel=self.channel_id, blocks=message, text=self.text_fallback, unfurl_links=False, unfurl_media=False)
return response['ts']
def _post_thread_message(self, result, message, message_ts):
if result.failed or result.skipped:
self.client.chat_postMessage(channel=self.channel_id, attachments=message, text=self.text_fallback, thread_ts=message_ts)
def _update_principal_message(self, result, message_timestamp, message: str):
if not result.parent:
self.client.chat_update(channel=self.channel_id, blocks=message, text=self.text_fallback, ts=message_timestamp)
def _build_principal_message(self, result, application, environment, executions, success_executions, failed_executions, skipped_executions):
'''
Builds the main message block
'''
if result.passed == False and result.failed == False:
start_suite_message[0]['text']['text'] = f'Aplicación en prueba: {application}'
if self.frontend_test:
start_suite_message[7]['text']['text'] = f'*CICD*: *<{self.cicd_url}|{self.cicd_id}>* || *BrowserStack*: *<{self.devicefarm_url}|{self.cicd_id}>*'
else:
start_suite_message[7]['text']['text'] = f'*CICD*: *<{self.cicd_url}|{self.cicd_id}>*'
return start_suite_message
else:
result_status = result.status
if result.passed:
result_icon = ":large_green_circle:"
elif result.failed:
result_icon = ":red_circle:"
else:
result_icon = ":white_circle:"
principal_block[0]['text']['text'] = f'Aplicación en prueba: {application}'
principal_block[1]['text']['text'] = f'*Branch*: {self.branch} || *Enterno*: {environment}'
principal_block[4]['text']['text'] = f'{result_icon} *{result_status}*'
principal_block[7]['fields'][0]['text'] = f'*Pruebas Ejecutadas:*\n{executions}'
principal_block[7]['fields'][1]['text'] = f'*Probado con éxito:*\n{success_executions}'
principal_block[8]['fields'][0]['text'] = f'*Probado con error:*\n{failed_executions}'
principal_block[8]['fields'][1]['text'] = f'*Pruebas salteadas:*\n{skipped_executions}'
if self.frontend_test:
principal_block[11]['text']['text'] = f'*CICD*: *<{self.cicd_url}|{self.cicd_id}>* || *BrowserStack*: *<{self.devicefarm_url}|{self.cicd_id}>*'
else:
principal_block[11]['text']['text'] = f'*CICD*: *<{self.cicd_url}|{self.cicd_id}>*'
return principal_block
def _build_thread_message(self, result, attachment_color):
tread_error_message[0]['color'] = f'{attachment_color}'
tread_error_message[0]['blocks'][0]['text']['text'] = f'{result.name}'
tread_error_message[0]['blocks'][3]['text']['text'] = f'{result.message}'
return tread_error_message
def _attachment_color(self, result):
color = None
if result.passed:
color = "1abf00"
elif result.failed:
color = "ff4646"
elif result.skipped:
color = "eddd00"
return color- const_messages.py
principal_block = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Aplicación en prueba: OneApp Mobile",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": " || *Branch*: main || *Entorno*: qa"
}
},
{
"type": "divider"
},
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Estado de la prueba:",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": ":red_circle: FAIL"
}
},
{
"type": "divider"
},
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Resumen:",
"emoji": True
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": "*Pruebas Ejecutadas:*\n17"
},
{
"type": "mrkdwn",
"text": "*Probado con éxito:*\n15"
}
]
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": "*Probado con error:*\n2"
},
{
"type": "mrkdwn",
"text": "*Pruebas salteadas:*\n0"
}
]
},
{
"type": "divider"
},
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Siga la ejecución:",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*CICD*: *<https://google.com|0Ghry48>* || *BrowserStack*: *<https://google.com|0Ghry48>*"
}
}
]
start_suite_message = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Aplicación en prueba: OneApp Mobile",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*Branch*: main || *Entorno*: qa"
}
},
{
"type": "divider"
},
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Estado de la prueba",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": ":slack_load: *En Prueba*"
}
},
{
"type": "divider"
},
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Siga la ejecución:",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*CICD*: *<https://google.com|0Ghry48>* || *BrowserStack*: *<https://google.com|0Ghry48>*"
}
}
]
tread_error_message = [
{
"color": "ff4646",
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Cenário: XPTO"
}
},
{
"type": "divider"
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*Mensaje*"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "_Erro XPTO_"
}
}
]
}
]- pyproject.tml
[tool.poetry]
name = "robotframework-slack"
packages = [
{include = "RobotSlackNotification"}
]
[tool.poetry.dependencies]
python = "^3.9"
slack-sdk = "3.21.2"
robotframework = "^5.0.1"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
Vamos entender alguns pontos importantes do código acima.
O arquivo “__init__.py” é quem possui toda a lógica do ouvinte. Abaixo, uma breve explicação dos argumentos do método “__init__” e de cada método que compõe a classe da biblioteca:
- __init__():
- slack_token: É o token de autenticação do Slack, que permite o acesso à API do Slack para enviar mensagens.
- channel_id: É o ID do canal do Slack para onde as mensagens serão enviadas.
- application: É o nome da aplicação que está sendo testada.
- environment: É o ambiente em que os testes estão sendo executados.
- branch: É o nome da branch do código fonte em que os testes estão sendo executados. Ajuda a rastrear os resultados dos testes executados em diferentes branches.
- cicd_url: É a URL do sistema de integração contínua/entrega contínua (CI/CD) que está executando os testes.
- cicd_id: É um identificador único para o processo de CI/CD em execução.
- devicefarm_url: É a URL para o Devicefarm onde os testes estão sendo executados.
- frontend_test: É um parâmetro opcional que indica se os testes são de frontend. Se for definido como True, o devicefarom_url será exibido no corpo da mensagem do Slack.
- start_suite(self, data, result): Esse método é chamado no início de cada suite de testes. Ele constrói e envia uma mensagem principal para o Slack, informando o início da suíte de testes. A mensagem inclui detalhes como a aplicação em teste, o ambiente, o número total de testes, os testes bem-sucedidos, os testes com falha e os testes pulados.
- end_suite(self, data, result): Esse método é chamado no final de cada suíte de testes. Ele atualiza a mensagem principal enviada anteriormente, adicionando os resultados finais da suíte de testes, como o número total de testes, os testes bem-sucedidos, os testes com falha e os testes pulados.
- end_test(self, data, result): Esse método é chamado no final de cada cenário de teste. Ele constrói uma mensagem de thread que é enviada para o Slack, informando o resultado da execução do cenário. A cor da mensagem de thread varia dependendo do resultado do teste (bem-sucedido, com falha ou pulado).
- _post_principal_message(self, result, message: str): Esse método é responsável por enviar a mensagem principal para o Slack. Ele utiliza a API do Slack para postar a mensagem no canal especificado. A mensagem é composta por blocos de texto formatados de acordo com as informações fornecidas, como a aplicação em teste, o ambiente e os resultados da suíte de testes.
- _post_thread_message(self, result, message, message_ts): Esse método é responsável por enviar a mensagem de thread para o Slack. Ele é chamado apenas para os testes que falharam ou foram pulados. A mensagem de thread é anexada à mensagem principal anteriormente enviada e é exibida como uma resposta em um thread no Slack.
- _update_principal_message(self, result, message_timestamp, message: str): Esse método é responsável por atualizar a mensagem principal no Slack. Ele é chamado no final da suíte de testes para atualizar a mensagem principal com os resultados finais da suíte de testes.
O arquivo “const_messages.py” servirá de base para a construção dos blocos mensagens que serão enviadas para o Slack. Esses modelos são atualizados durante a execução dos testes, com os dados coletados pelos métodos ouvintes do Robot Framework (start_suite, end_suite e end_test) e enviados para o slack.
A estrutura do projeto ficou bem simples:
.
├───RobotSlackNotification
│ │ const_messages.py
│ │ __init__.py
│ pyproject.tomlApós construído, bastou subir para o Github e realizar a instalação desse pacote via Poetry nos projetos que quero utilizar essa biblioteca.
Após instalar a biblioteca no projeto, o comando de execução no Robot Framework fica um pouco extenso, já que cada argumento do método __init__() se torna uma argumento do comando de execução. Dessa forma, o comando fica dessa forma:
robot --norpa -d report/log \
--listener "RobotSlackNotification;{slack_token};{channel_id};{application};{environment};{branch};{cicd_url};{cicd_id};https://app-automate.browserstack.com/dashboard/v2/search?query={runner_id}&type=builds;{frontend_test}" \ .E como ficam as mensagens no Slack?
Alguns prints das mensagens enviadas ao slack, em tempo real de execução de teste, ao Slack.
OBS 1.: Algumas informações não estão visíveis por sigilo ✌
OBS 2: Todas as mensagens estão em espanhol por ser a língua predominante da empresa onde trabalho. ✌
Nas imagens abaixo, podemos ver os 3 estados principais: Em andamento, Falha e Sucesso. Além dos dados sumarizados da execução, que são atualizados em tempo real.
E abaixo, um exemplo de como é construído a Thread caso algum cenário apresente falha ou seja pulado.
E foi dessa forma que criei minha biblioteca Listener para envio de notificações para o Slack.
—
E ai, curtiu? O que achou da ideia?
Gostaria que essa biblioteca fosse para o PyPi? Diz ai nos comentários.
Não esquece de compartilhar esse post com seus amigos que curtem automação e o Robot Framework. Isso ajuda demais no crescimento desse pequeno projeto.
Um grande abraço!
Valeu
\o/