()
| 11 | |
| 12 | |
| 13 | async def main(): |
| 14 | client = SocketModeClient( |
| 15 | app_token=os.environ.get("SLACK_SDK_TEST_SOCKET_MODE_APP_TOKEN"), |
| 16 | web_client=AsyncWebClient(token=os.environ.get("SLACK_SDK_TEST_SOCKET_MODE_BOT_TOKEN")), |
| 17 | trace_enabled=True, |
| 18 | ) |
| 19 | |
| 20 | async def process(client: SocketModeClient, req: SocketModeRequest): |
| 21 | if req.type == "events_api": |
| 22 | response = SocketModeResponse(envelope_id=req.envelope_id) |
| 23 | await client.send_socket_mode_response(response) |
| 24 | if req.payload["event"]["type"] == "message": |
| 25 | await client.web_client.reactions_add( |
| 26 | name="eyes", |
| 27 | channel=req.payload["event"]["channel"], |
| 28 | timestamp=req.payload["event"]["ts"], |
| 29 | ) |
| 30 | |
| 31 | client.socket_mode_request_listeners.append(process) |
| 32 | await client.connect() |
| 33 | await asyncio.sleep(float("inf")) |
| 34 | |
| 35 | |
| 36 | asyncio.run(main()) |
no test coverage detected