[Python, aioresponses] pytest 테스트 코드 작성시 외부 api Mocking 하기.

 

테스트에 대한 개념과 코드 작성을 현재 배워가는 단계라 미숙한 부분이 많아 기록용으로 정리하고자 한다!

 

테스트 코드를 짜놓지 않으면 그동안 테스트를 어떻게 했었는지 기억을 더듬어본다..

 

기존에는 api 코드를 작성하고 api 테스트하는 툴, DB 접속 클라이언트 툴 창을 일일히 띄워놓고

디비에 값이 제대로 저장이 되나, 값을 제대로 읽어오나, CRUD를 손으로 테스트하고,

다양한 케이스 별 테스트를 매번 수동으로 반복하기가 번거롭고 쉽지도 않고

테스트를 하는 날에 따라서 정리가 안되어 놓칠 수 있는 부분이 많았다.

 

현재 회사에서 테스트 코드 작성을 처음 접하게 되었는데,

pytest, unittest를 이용하여 endpoint 테스트,

또 Serializer를 작성했다면 각 serializer의 기능들을 최대한 잘게 쪼개서 테스트 코드를 작성하는 연습을 하고 있다.

 

또 해당 function에 다양한 상황의 변수를 대입하여

(예를 들면 에러를 뱉는 상황, 응답값이 단일값, 여러값인 상황 등등....) 테스트로 최대한 다양한 상황을 커버할 수 있는 연습도 같이!

 

이렇게 한번 테스트 코드를 꼼꼼히 작성해놓으면, 수동으로 다양한 케이스를 커버하는 작업을 하지 않고 코드로 자동화시킬 수 있는 것이다.

그만큼 테스트 코드 작성도 개발 코드를 작성하는 것 만큼(?) 중요하고 꼼꼼하게 작업해야 한다.

이렇게 테스트 코드의 중요성이 대두되는 만큼 TDD의 방법론도 많이 사용하고 있다고 한다.

 

* TDD(Test-Driven-Development) : 테스트 주도 개발의 개념도 테스트 코드를 먼저 작성하고, 후에 테스트 코드에 맞게 개발코드를 작성하는 방법론이다. 

 

 

무튼,,

그렇게 테스트 코드를 작성하면서 Mocking과 aioresponses를 써야하는 상황에 직면하게 되었다.

처음엔 Mocking의 개념 자체도 생소해서 많이 해맸는데, 코드를 작성하면서 동료분들의 자세한 설명과 써야하는 상황을 직접 직면해보니

나름대로 개념이 잡혔다.

 

---

Mocking이란 쉽게 말해서, 테스트 scope안에서 가상으로 특정 funtion이나 api 요청에 대한 응답값을 설정할 수 있다. 다시말해서 function이나 api요청에 예상되는 응답값을 테스트 범위 내에서 가짜로 박아놓는(?) 작업이다. 그래서 api Mocking을 하게 되면, 테스트 실행시 실제로 해당 api에 요청을 보내지 않고도 응답값을 가상으로 설정할 수 있다.

---

 

 

 

그리고 aioresponses는 위 설명처럼 테스트 코드 작성시 외부 api 서비스에 대한 요청이 필요할 때

응답값을 가상으로 Mocking 시킬 수 있는 파이썬의 aiohttp 패키지에 있는 helper이다.

 

 

 

 

아래 예시 코드를 보자. (github 문서 참조)

import asyncio
import aiohttp
from aioresponses import aioresponses

def test_ctx():
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    with aioresponses() as m:
        m.get('http://test.example.com', payload=dict(foo='bar'))

        resp = loop.run_until_complete(session.get('http://test.example.com'))
        data = loop.run_until_complete(resp.json())

        assert dict(foo='bar') == data

 

context manager 방식으로 작성한 코드여서

with aioresponses() as m: ~ 이하의 indentation 범위에서 Mocking이 유효하다.

 

m.get('http://test.example.com', payload=dict(foo='bar'))

 

test.example.com에 get 요청을 했을 때, payload 필드에 가상의 응답값을 위처럼 설정할 수 있다.

 

 

resp = loop.run_until_complete(session.get('http://test.example.com')) 
data = loop.run_until_complete(resp.json())

assert dict(foo='bar') == data

 

그렇다면 실제로 Mocking이 되었는지 테스트 코드를 작성할 수 있다.

session.get()으로 위에서 설정한 URL로 get 요청을 보내 응답값을 data에 받아온다.

 

우리는 위의 응답값 data와 payload로 Mocking시킨 데이터 dict(foo='bar')가 같다고 예상할 수 있다 !

 

 

 

 

아래는 post 요청에 대한 Mocking 코드 예시이다.

import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses()
def test_http_headers(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    m.post(
        'http://example.com',
        payload=dict(),
        headers=dict(connection='keep-alive'),
    )

    resp = loop.run_until_complete(session.post('http://example.com'))

    # note that we pass 'connection' but get 'Connection' (capitalized)
    # under the neath `multidict` is used to work with HTTP headers
    assert resp.headers['Connection'] == 'keep-alive'

 

post의 경우 headers 설정을 추가할 수 있다.

필자의 경우 post 요청시, 요청 body에 따라서 다른 Mocking을 할 수 있는지 찾아봤는데 body argument는 별도로 없는 것 같다..

 

 

 

 

---

위의 예제코드는 aioresponses 공식 문서에 있는 예시 코드이고,

자 이제 필자가 작성한 코드를 보자.

직접 활용하면서 몇번(?)의 삽질을 경험하면서 귀중한 사실들을 알게 되었다.

 

    @pytest.mark.asyncio
    async def test_new_data_save(self, dynamodb_proc):
        data = {
            'user_id': uuid.uuid1().hex,
            'ip_addr': constants.MOCKING_DATA[0]['ip']
        }

        mocking_data = constants.MOCKING_DATA[0]

        # m.get(url..)을 여러번 쓰면 작성한 횟수만큼 외부 요청을 여러번 Mocking할 수 있다.
        # aioresponses쓰면 외부의 모든 요청을 불가능하게 막고, mocking한 URL만 요청 가능하게 함.
        with aioresponses(passthrough=[f'http://localhost:{dynamodb_proc.port}/']) as m:
            url = f'https://api.ipdata.co/{data["ip_addr"]}?api-key={config.API_KEY}'\
                '&fields=ip,city,region,country_name,continent_name,latitude,longitude'
            m.get(url, payload=mocking_data, status=200)
            serializer = SaveSerializer(ip_addr=data['ip_addr'], user_id=data['user_id'])

            try:
                await serializer.serialize()
            except exceptions.AttributeNoneError as e:
                assert e.msg == "continent_code, continent_name, country_code, country_name attributes cannot be None."

            user_ip_data = await self._get_from_user_ip_database(ip_addr=data['ip_addr'], user_id=data['user_id'])

            ip_data = await IPData.get(hash_key=data['ip_addr'])

            assert ip_data is not None
            assert user_ip_data is not None
            assert len(user_ip_data) > 0

            for db_data in user_ip_data:
                assert db_data['user_id'] == data['user_id']
                assert db_data['ip_addr'] == data['ip_addr']

            assert ip_data.ip_addr == data['ip_addr']

            assert mocking_data == ip_data.as_dict()

 

 

위 테스트는 새로운 데이터를 2개의 DynamoDB 테이블에 저장하기 위한 코드이다.

새로운 데이터를 ipdata api에 요청하여 응답값을 받아와 저장하는 코드가 개발 코드의 로직인데,

DB에 예상한 값대로 잘 들어가는지, api 응답 데이터를 예상한대로 parsing이 되는지의 테스트가 필요하였다.

 

이 과정에서 ipdata api 요청에 대한 Mocking이 필요하여 aioresponses를 Context Manager 방식으로 작성하였다.

 

 

그러면 한줄 한줄 코드를 보자.

 

with aioresponses(passthrough=[f'http://localhost:{dynamodb_proc.port}/']) as m:
            url = f'https://api.ipdata.co/{data["ip_addr"]}?api-key={settings.API_KEY}'\
                '&fields=ip,city,region,country_name,continent_name,latitude,longitude'
            m.get(url, payload=mocking_data, status=200)

 

 

aioresponses를 사용한 핵심코드이다.

passthrough라는 argument는 인자로 받은 URL을 '허용'시켜준다는 의미이다.

aioresponses를 사용하면, aioresponses에서 Mocking한 url만 요청이 가능하며,

그 url 이외의 모든 요청을 불가능하게 막는다.

 

Mocking scope 내의 테스트 코드에서 DynamoDB에 저장하는 요청이 있었어서,

위처럼 dynamodb URL을 passthrough에 등록하지 않으면 아래의 ClientConnectionError 오류가 난다.

aiohttp.client_exceptions.ClientConnectionError: Connection refused: POST http://localhost:12345/

위처럼, Mocking URL이외의 요청이 aioresponses Mocking scope 내에서 필요할 경우,

passthrough arguments에 list로 다른 외부 URL을 등록해주면 된다.

 

mocking_data값이 URL 요청에 대한 응답이 오도록 m.get()으로 Mocking해준 코드이다.

 

 

나머지 그 아래의 코드는,

api 응답값을 Model형태에 맞게 parsing하고 DB에 저장하는 로직의 function을 호출한뒤,

DB에 예상한 값이 제대로 들어갔는지,

Mocking한 데이터를 파싱한 값이 DB에서 꺼낸값과 같은지를 비교해주는 코드이다.

 

 

 

 

 

 

https://github.com/pnuckowski/aioresponses

pnuckowski/aioresponses