작년 11월, 구글 번역 서비스가 신경망 학습을작년 11월, 구글 번역 서비스가 신경망 학습을 적용해서 개편되었다. (관련기사) 벌써 서너달이 지난 일이지만, 현재까지도 도움을 많이 받고 있다.

google-translate-test

너무 잘되서 놀람

번역을 시켜보니 품질이 너무 괜찮고 재밌어서 당시에 바로 봇으로 만들었다. 그리고 다른 사람들도 각자의 슬랙에서 쓸 수 있도록 공개했다. 슬랙 봇 개발과 테스트 작성이 처음인 누군가에게는 도움이 되리라고 믿는다.

완성본은 이곳에 있다.

요구사항

기능

다음과 같은 유저 기능이 필요하다.

  1. 구글 번역 API를 이용한 슬랙 봇을 만든다.
  2. 타겟 언어를 지정해서 번역을 시킨다. (기본값 존재)
  3. 가능한 타겟 언어를 조회할 수 있다.

끝이다. 기능은 매우 단순하다.

추가 요구사항

그러나 공개를 위해서 다음과 같은 요구사항을 추가했다.

  1. 모든 기능의 테스트 작성
  2. CI 구축을 통한 테스트 자동화
  3. 테스트 커버리지 측정

사실 기능 개발보다는 추가 요구사항을 만족시키는 것이 더 어려웠다.

기능 요구사항

전제 및 사용 방법

  • gtbot이라는 이름으로 슬랙 앱을 등록한다.
  • 슬랙 채널에서 @gtbot 번역할 문장으로 사용한다.

의존성

Slack Bot Token 받기

  • 봇을 등록할 슬랙 팀에서 App & Integrations 메뉴를 선택하고 Slack API 페이지로 이동한다.
  • 왼쪽의 사이드 메뉴에서 App features -> Bot users 메뉴를 선택한다.

create-bot-user

  • creating a new bot user를 눌러 봇을 생성하는 화면으로 이동한다.
  • 이름을 gtbot으로 정한다.
  • Add bot intergration 버튼을 눌러 설정을 입력하고 토큰을 받아 놓는다.

Google Translation API Token 받기

** 구글 번역 API는 유료입니다! 주의하세요! **

개발

자 이제 봇을 만들어 보자.

봇 생성

# bot.py

import os
from slacker import Slacker

def create_bot():
    slack_token = os.environ.get('GTBOT_SLACK_TOKEN')
    google_token = os.environ.get('GTBOT_GOOGLE_TOKEN')
    return Bot(Slacker(slack_token), Translator(google_token))

def run():
    bot = create_bot()
    bot.run_loop()

if __name__ == '__main__':
    run()

그냥 python bot.py로 봇을 실행시킬 생각이기 때문에, 다음과 같이 진입점을 구성했다. 봇이 생성되면 무한 루프를 돌면서 요청을 처리한다.

메세지 송수신

class Bot:
    def run_loop(self):
        print('Start event loop...')

        while True:
            ch, msg, target = self._read()
            if msg == '/lang':
                msg = self._translator.availables()
            elif msg == '/setdefault':
                msg = self._set_default(target)
            else:
                msg = self._translator.translate(msg, target=target)

            self._send(ch, msg)

    def _read(self):
        while True:
            event = json.loads(self._socket.recv())
            if 'bot_id' in event and event.get('username') != 'testuser':
                continue

            ch, msg, target = self._parse(event)
            if not ch or not msg:
                continue

            break

        return ch, msg, target

    def _send(self, ch, msg):
        self._slacker.chat.post_message(ch, msg, as_user=True)

메세지 수신을 위한 루프가 2개 있는데,

  • _read: 하나의 메세지를 기다리는 루프
  • run_loop: 영원히 메세지를 처리하기 위한 루프

이고, 송신은 _send로 처리한다.

현재 /lang이나 /setdefault 같은 명령어는 다음과 같이 사용한다.

@gtbot /setdefault en, @gtbot /lang

메세지 파싱

class Bot:
    def _parse(self, event):
        if event['type'] != 'message' or self._gtbot_id not in event['text']:
            return None, None, None

        text = event['text'].replace(self._gtbot_id, '').strip()
        target = self._default_target

        if text.startswith('/target'):
            text = text.replace('/target', '').strip().split()
            target, text = text[0], ' '.join(text[1:])

        if text.startswith('/setdefault'):
            text, target = text.split()

        return event['channel'], text, target

이벤트 타입이 텍스트가 아니거나, @gtbot으로 직접 호출하지 않은 경우는 무시한다.

구글 번역 모듈

requests로 API를 호출한다. 구글 번역 API의 자세한 스펙은 공식 문서를 참고한다.

추가 요구사항

테스트

의존성

참고한 프로젝트

기능 테스트를 실행하기 위한 실제 슬랙 환경을 구축하는 것이 생각보다 까다로웠다. 참고할 자료를 찾다 다음과 같은 프로젝트를 발견했다.

작성 개요

작성한 기능 테스트는 다음과 같다.

  1. 봇과 각 구성 요소의 생성
  2. 번역
  3. 가능한 타겟 언어 리스트업
  4. 타겟 언어를 지정해서 번역
  5. 기본 타겟 언어 변경

1번 테스트는 인스턴스의 타입을 확인하는 방법을 쓰면 될 것 같다. 2-5번 테스트는 아마 다음과 같은 과정을 거칠 것이다.

  1. 봇 프로세스를 생성
  2. 테스트 채널을 만든다.
  3. 테스트 유저의 이벤트 루프를 생성한다.
  4. 테스트 유저가 봇에게 기능을 요청한다.
  5. 응답이 맞는지 확인한다.

테스트 유저 Slack Token 받기

테스트 유저가 필요하므로 추가 토큰이 필요하다. 레거시 토큰을 받는 메뉴로 이동한다. 원하는 팀 옆의 Create Token 버튼을 눌러 토큰을 생성하고 잘 저장한다.

테스트 케이스 작성

대표적으로 번역 기능의 테스트 케이스를 작성해본다.

일반적인 테스트 케이스와 다른 점은 assert문을 직접 사용하지 않는다는 것이다. 대신 불리언을 반환하는 assert 함수를 받고, False를 리턴하면 AssertoinError를 낸다. 지정한 시간 내로 응답이 오지 않아도 에러를 낸다.

@pytest.mark.usefixtures('run_bot_process')
@pytest.mark.usefixtures('run_testuser_msg_loop')
@pytest.mark.usefixtures('join_testchannel')
class TestBot:
    def test_translate(self, bot_id, testuser_slacker, testchannel_id, bot_reply):
        self._send_msg('안녕', bot_id, testuser_slacker, testchannel_id)

        def assert_func(reply):
            return 'Hello' == reply
        self._wait_reply(bot_reply, assert_func)

def _wait_reply(self, bot_reply, assert_func):
    has_reply = False

    for _ in range(10):
        time.sleep(2)

        if bot_reply:
            has_reply = True

            if assert_func(bot_reply[0]):
                del bot_reply[:]  # remove all element in bot_reply
                break  # assert successfully, then break
            else:
                raise AssertionError('setting default language failed')

    if not has_reply:
        raise AssertionError('no expected message')

‘안녕’을 번역 요청하면 ‘Hello’라는 응답이 오는지 확인하는 테스트 케이스라는 것을 알 수 있다.

봇 프로세스 생성

위의 테스트 케이스에 보면, @pytest.mark.usefixtures('run_bot_process')라는 픽스쳐가 봇 프로세스를 생성하는 것을 짐작할 수 있는데, 동작은 다음과 같다.

@pytest.fixture(scope='session')
def run_bot_process(bot_id, testuser_slacker):
    sp = subprocess.Popen('python bot.py', shell=True)

    # waiting for bot online
    for _ in range(10):
        time.sleep(2)

        bot_id = bot_id[2:-1]
        resp = testuser_slacker.users.get_presence(bot_id)
        if resp.body['presence'] == testuser_slacker.presence.ACTIVE:
            break
    else:
        raise AssertionError('bot is offline')

    yield sp
    sp.terminate()

subprocess.Popen으로 프로세스를 생성하고, 연동된 슬랙 팀에서 봇이 온라인 상태인지를 체크한다.

테스트 유저의 채널 참석

테스트 유저가 봇에게 말을 걸기 위해서 슬랙에 테스트 채널을 하나 만들고(여기서는 test_bot이다) 테스트 유저를 입장시킨다.

@pytest.fixture(scope='session')
def join_testchannel(testuser_slacker):
    test_token = os.environ.get('GTBOT_SLACK_TOKEN_TEST')
    testuser_slacker = slacker.Slacker(test_token)

    resp = testuser_slacker.channels.join('test_bot')
    return resp

테스트 유저의 이벤트 루프 생성

테스트 유저가 봇이 응답하는지 확인 할 수 있도록, 별도의 쓰레드에 이벤트 루프를 만들고 기다린다.

@pytest.fixture(scope='session')
def run_testuser_msg_loop(websocket, bot_reply, bot_id):
    def loop(websocket):
        while True:
            while True:
                event = json.loads(websocket.recv())
                # some event has no user
                user_id = '<@{}>'.format(event.get('user'))

                if event['type'] == 'message' and user_id == bot_id:
                    bot_reply.append(event['text'])

    start_new_thread(loop, (websocket,))

CI 구축

이제 매 커밋마다 기능이 제대로 돌아가는지 확인할 수 있도록 CI를 구축하고 자동 테스트를 시켜놓자.

Travis CI

Github 저장소 연동

Travis CI 페이지로 들어가서 Github으로 로그인하면 Github 저장소를 쉽게 연동 가능하다. 왼쪽 My Repositories 메뉴의 왼쪽 ‘+’ 버튼을 눌러서, 원하는 저장소의 스위치를 켜준다.

travis-new-repo

설정파일 작성

이제 프로젝트 루트 디렉토리에 트래비스 설정 파일을 .travis.yml이라는 이름으로 작성한다.

language: python
python:
    '3.5'
install:
    pip install -r requirements.txt
script:
    pytest --cov
after_success:
    bash <(curl -s https://codecov.io/bash)

--cov 옵션으로 테스트 커버리지를 측정한다. 이 때, 설정 파일이 필요하다. 프로젝트 루트에 .coverage를 만들고 커버리지를 측정할 대상이 되는 소스코드를 지정해준다.

[run]
source =
    bot.py

사실 유닛테스트가 없기 때문에 커버리지는 별로 의미는 없지만 그냥 재미로 붙여봤다.

마지막 after_success항목에 Codecov 서비스로 커버리지 측정 결과를 전송하는 역할을 한다.

Codecov

일단 Codecov 페이지로 들어가서 Github으로 로그인한다. Add new repository 버튼을 누르고 원하는 저장소를 선택한다.

자, 이제 저장소에 코드를 푸시하고 테스트가 잘 돌아가는지 확인해보자. 그리고 Codecov로 날아온 리포트를 확인해보자!

후기

재밌고 유익한 프로젝트였다. 특히, 다른 사람에게 공개할 소프트웨어를 만들기 위해서는 (별로 쓰는 사람은 없어도 ㅋㅋ) 훨씬 더 심혈을 기울여야 한다는 것을 알게 되었다.

또 프로젝트의 크기가 작은 편이라 일주일간 퇴근시간을 이용해서 만들 수 있었는데, 토이 프로젝트의 크기로서 적당했다. 너무 크거나 오래 걸리면 하다 지쳐서 포기하게 되고, 그럴수록 좌절감만 쌓이는데, 이 프로젝트는 빠르게 개발하고 완성해서 성취감을 얻을 수 있었다는 점이 가장 좋은 점이었다.

기록을 정리하면서 새 프로젝트를 진행하고 싶다는 생각이 들었다. 이번엔 더 나에게도 다른 사람에게도 더 도움되는 소프트웨어를 만들어보고 싶다.