aioresponses에 대한 본격적인 내용은 서론 아래에 있습니다

 

목차

- @aioresponses() 데코레이터 사용

- Context Manager를 활용한 aioresponses()

- 동일 URL 반복 호출

- 특정 URL만 실제 api로의 요청을 허용하는 방법

 


 

 

 

파이썬 서버 테스트 과정에서 공부한 외부 api mock out 방법에 대해 기록하고자 한다

 

 

현 회사에서 최근 서버 개발을 위한 unit test를 작성하는 방법에 대해 새롭게 익히고 있다.

기존에는 postman을 통해 e2e test를 중심으로 개발하였는데, e2e test는 사용자 입장에서의 테스트이기 때문에 개발 디버깅용으로는 비효율적인 방식이었다. 

 

일반적으로 서버가 MVC 방식으로 모듈화 되어있다고 했을 때, 각 모듈이 어떻게 동작하는지를 먼저 테스트 해야한다.

이때 사용하는 것이 unit test(단위 테스트)다. 

파이썬 unit test 툴로는 가장 유명한 것이 Pytest와 Unitttest 두 가지가 있는데, 필자는 pytest를 중심으로 공부하고 있다.

 

 

최근 크롤링 등의 이유로 외부 api를 많이 호출하면서 단위 테스트 과정에서 문제가 되었다.

 

예를 들어 크롤링해야하는 웹페이지는 로그인을 해야지만 접근할 수 있다면 테스트를 할 때 authorization 헤더에 매번 올바른 토큰값을 넣어줘야 한다.

이를 위해 주기적으로 로그인을 한 후 토큰값을 가져와 헤더에 넣어주는 절차를 거쳐야하는데, 비용 측면에서 큰 낭비가 될 것이다.

 

이 때 사용할 수 있는 방법은 외부 페이지를 mocking 하는 것이다.

mock은 서로 의존성이 강한 객체가 있을 때 가짜 객체를 만들어 사용하는 것을 말한다

 

 

그렇다면 외부 api를 호출할 때 어떻게 mocking한 페이지를 가져올까? 이 때 사용하는 것이 aioresponses 라이브러리다

 

 

 

 


 

 

 

 

aioresponses는  aiohttp request를 테스트할 때 사용할 수 있는 helper다.

 

requests 모듈을 테스트할 때 사용하는 툴은 httpretty, responses, requests-mock 등 많이 제공되고 있지만 aiohttp와 같이 비동기 방식이 적용되면 테스트 난이도가 올라가게 된다. 이때 사용할 수 있는것이 aioresponses다

 

aioresponses는 aiohttp session을 통해 외부 api로 request를 보낼때, mocking한 가상의 response를 대신 받을 수 있도록 한다.

 

aioresponses를 사용하는 방법에 대해 알아보자

 

 

 


 

@aioresponses() 데코레이터 사용

 

다음 코드를 보자

import aiohttp
import asyncio
from aioresponses import aioresponses

TEST_URL = "https://test.co.kr"

@aioresponses()
async def test(mocked):
    # 가상의 응답 설정
    mocked.get(TEST_URL, status=200, payload=dict(test="Success"))
    
    # 외부 api 호출
    session = aiohttp.ClientSession()
    async with session.get(TEST_URL) as response:
        print(await response.json())
    await session.close()


if __name__ == '__main__':
    asyncio.run(test())

 

결과

{'test': 'Success'}

 

 

@aioresponses() 데코레이터를 단 함수는 첫번째 인자로 aioresponses 객체를 받는다.

이 객체를 사용하면 어떤 api 호출에 어떤 응답을 줄 것인지 지정할 수 있다.

 

mocked 변수에 받은 aioreponses 객체를 사용하는 코드를 살펴보자

 

mocked.get(TEST_URL, status=200, payload=dict(test="Success"))

 

aioresponses에는 get, post, put, delete 등 http method를 지원한다.

인자로는 호출하고자 하는 api url를 전달해야하며, status, payload, body 등 다양한 응답을 지정할 수 있다.

 

payload의 인자로는 dict, str, list 등을 사용할 수 있지만 body의 경우 str만 전달이 가능하다

만약 json을 응답으로 전달한다면 payload가 좋을 것이고, html을 전달한다면 body가 나은 선택일 수 있다.

 

이외에 headers, exception, status 등도 직접 설정이 가능하다. 

 

 

 


 

Context Manager를 활용한 aioresponses

 

위 코드에서 데코레이터를 활용한 방식을 소개했다면 이번에는 context manager를 사용한 방식을 소개하려 한다

 

여기서 context manager란 with 구문을 사용하여 특정 자원을 정확한 시간동안 사용하고 종료할 수 있도록 하는 것을 말한다.

즉, with 구문 아래에 들여쓰기가 되어있는 코드에만 할당한 자원을 사용할 수 있도록 하는 것을 말한다.

위 코드에서 이미 aiohttp session을 사용하는데 활용했기 때문에 여기서는 간단하게만 설명하였다

 

 

다음 코드를 살펴보자

 

TEST_URL = "https://test.co.kr"

async def test():
    # context manager를 사용한 aioresponses
    with aioresponses() as mocked:
        mocked.get(TEST_URL, status=200, payload=dict(test="Success"))

        session = aiohttp.ClientSession()
        async with session.get(TEST_URL) as response:
            print(await response.json())
        await session.close()

 

결과

{'test': 'Success'}

 

위 코드를 보면 aioresponses 메서드에서 해당 객체를 mocked 변수에 받아 사용하고 있다.

아래에 코드는 데코레이터를 사용한 코드와 동일하다.

 

이 방법은 외부 api를 사용하는 부분이 전체 코드의 일부분이고, 복잡한 코드로 인해 에러가 우려되거나 메모리 누수(resource leak) 우려가 있는 경우 사용할 수 있다.

데코레이터는 함수 전체가 끝날 때까지 aioresponses 객체를 계속 유지한다.

때문에 외부 api 호출이 주요 기능인 경우가 아니라면 위와 같이 context manager를 사용해서 일정 시간동안만 리소스를 사용하고 안전하게 닫는 것이 더 좋은 방법일 수 있다.

 

 

 


 

동일 URL 반복 호출

 

이렇게 설정한 가상의 응답은 기본적으로 1회만 사용이 가능하다. 

아래의 예시 코드를 보자

 

TEST_URL = "https://test.co.kr"

@aioresponses()
async def test(mocked):
    mocked.get(TEST_URL, status=404)	# 첫번째 응답
    mocked.get(TEST_URL, status=200)	# 두번째 응답

    session = aiohttp.ClientSession()
    async with session.get(TEST_URL) as response:
        print(response.status)
    async with session.get(TEST_URL) as response:
        print(response.status)
    await session.close()

 

결과

404
200

 

위 코드에서는 같은 url에 대해 mocking된 응답을 2개 만든 뒤 같은 url을 2번 호출하였다.

결과는 보다시피 첫번째 요청은 첫번째 응답을, 두번째 요청은 두번째 응답을 받았다.

 

만약 같은 url에 대해 다양한 응답을 받고 싶다면 위와 같은 방법을 사용하면 된다.

 

 

 

그렇다면 만약 같은 url을 여러 번 호출하고 계속 같은 응답을 받고 싶다면 어떻게 하면 될까?

이때 사용할 수 있는 파라미터가 repeat 이다. 

 

가령 url은 같지만 body를 바꿔가며 여러개의 응답을 전부 받아야하는 경우가 있을 수 있는데  repeat=False (default)로 되어있는 경우 Connection refused 에러가 발생하게 된다. 

내가 mocking 한 응답을 재사용한다는 의미로 repeat=True를 해주어야만 반복적인 요청이 가능하다.

 

예시 코드를 보자

 

TEST_URL = "https://test.co.kr"

@aioresponses()
async def test(mocked):
    mocked.get(TEST_URL, status=200, repeat=True)	# repeat 인자 설정

    session = aiohttp.ClientSession()
    async with session.get(TEST_URL) as response:
        print(response.status)
    async with session.get(TEST_URL) as response:
        print(response.status)
    await session.close()

 

결과

200
200

 

 

 


 

특정 URL만 실제 api로의 요청을 허용하는 방법

 

aioresponses는 외부 api를 호출하는 모든 요청을 통제하고 응답이 mocking된 URL 요청만 허용한다.

만약 mocking되지 않은 URL을 임의로 호출하면 aioresponses는 해당 요청을 차단하고 에러를 보낸다.

 

아래의 코드를 보자

 

REAL_URL = "https://real.url.com"

@aioresponses()
async def test(mocked):
    session = aiohttp.ClientSession()
    # mocking하지 않은 외부 api 호출
    async with session.get(REAL_URL) as response:	
        print(await response.json())
    await session.close()

 

결과

aiohttp.client_exceptions.ClientConnectionError: Connection refused: GET https://real.url.com

 

위와 같이 mocking되지 않은 api를 호출하면 해당 요청은 Connection refused 되었다는 에러메세지를 볼 수 있다.

 

 

경우에 따라, 일부 api는 mocking된 응답을 받고 일부는 외부에서 실제 데이터를 가져와야 할 수 있다.

그런 경우에 사용할 수 있는 것이 passthrough 파라미터다.

 

aioresponses 함수를 호출할 때, passthrought 파라미터에 특정 url 리스트를 인자로 전달하면 해당 url은 외부로 요청할 수 있도록 허용한다. 

 

아래의 코드를 보자

 

TEST_URL = "https://test.co.kr"
REAL_URL = "https://real.url.com"

# https://real.url.com은 예시를 위한 임의의 URL입니다.
@aioresponses(passthrough=[REAL_URL])
async def test(mocked):
    mocked.get(TEST_URL, status=200, payload=dict(test="Success"))

    session = aiohttp.ClientSession()
    # mocking된 요청
    async with session.get(TEST_URL) as response:
        print(await response.json())
        
    # 실제 외부 api로의 요청
    async with session.get(REAL_URL) as response:
        print(await response.json())
        
    await session.close()

 

결과

{'test': 'Success'}
{'real_url': 'Connected'}

 

위 코드는 mocking 된 TEST_URL과 외부로 실제 요청을 보낼 REAL_URL 두개에 각각 요청을 보내고 있다.

TEST_URL은 가상의 응답을 받아왔고, REAL_URL은 외부 서버에서 응답을 받아온 것을 확인할 수 있다.

 

 


 

 

본 포스트는 제가 공부하는 과정에서 작성하였습니다. 정정사항 댓글은 언제나 환영합니다!

반응형

윈도우를 사용하는 개발자들 중 wsl을 통해 리눅스를 설치해서 사용하는 경우가 있습니다.

간혹 윈도우에서 리눅스로 파일을 전달해야하는 경우가 있는데 2가지 방법을 쓸 수 있습니다.

 

첫째는 공유폴더를 만드는 방법이고

두번째는 scp(Secure copy protocol) 명령어를 이용하는 방법인데

 

개인적으로 공유폴더를 만드는 방법은 방화벽 설정과 유저 설정 등등 윈도우에서는 상당히 번거로웠습니다.

따라서 scp를 이용해 특정 파일 몇 개를 직접 전송하는 방법을 사용해보려 합니다.

 

참고 : https://baekh-93.tistory.com/50

 

필자는 윈도우 10과 리눅스 Ubuntu를 사용하였습니다.

 

 


 

 

우선 파일을 전송하기에 앞서 리눅스에서 몇가지 설정이 필요합니다.

 

ssh부터 설치해봅시다.

 

sudo apt-get install openssh-server

 

그리고 sshd-config에 두군데를 수정합니다.

에디터로 저는 vim을 사용하지만 nano 등 본인이 편하신 걸로 사용하시면 됩니다.

 

sudo vim /etc/ssh/sshd_config

 

기나긴 설정값들 사이에 두 부분을 다음과 같이 수정합니다.

 

#Port 22
-> Port 22
(주석을 해제합니다)
PasswordAuthentication no
-> PasswordAuthentication yes

 

첫 번째는 22번 포트를 여는 설정입니다. 이 부분을 수정해야 윈도우에서 22번 포트로 접근할 수 있습니다.

PasswordAuthentication은 리눅스 서버에 접속할 때 root 유저의 비밀번호를 사용하도록 허용하는 설정입니다.

기본값은 no이며 설정이 꺼져있으면 공개키를 사용해야합니다.

로컬에서만 돌릴 예정이므로 편하게 비밀번호를 사용합시다.

 

설정을 변경하였다면 저장합시다.

vim의 경우 :w! 을 먼저 입력해 readonly였던 파일을 수정 가능하게 변경해야합니다. !를 사용할 것이기 때문에 위에서 sudo를 사용한 것입니다.

 

설정을 변경했으니 ssh를 재실행합니다.

 

sudo /etc/init.d/ssh restart

 

22번 포트가 열렸는지 확인하고 싶다면 다음 명령어를 입력합니다.

 

netstat -ntl

 

위 사진을 확인하면 잘 열린 것을 볼 수 있습니다.

 

 

파일을 전송할 차례입니다. 윈도우 Powershell로 넘어옵시다.

명령어는 다음과 같습니다.

 

scp [전송할파일경로] [리눅스서버계정ID]@[리눅스서버계정IP]:[전송받을경로]

(예시)
scp .\test.txt root@10.0.0.1:\home\user01\share

 

여기서 확인해야할 사항은 서버계정 아이디와 IP입니다.

계정 아이디@IP는 우분투 터미널에 초록색 글자로 아이디@IP: 이렇게 쓰여있는걸 참고하셔도 됩니다.

 

@ 오른쪽 IP가 DESKTOP- 등으로 되어있다면 이를 입력했을 때 scp에서 접속이 어려운 경우가 있습니다.(저의 경우가 그랬네요)

 

이 경우 다음 명령어로 ip를 직접 확인해야합니다.

 

ifconfig

 

빨간색으로 가려준 부분에 적인 숫자가 ip입니다. 255.255.255.255 형식으로 적혀있습니다.

 

scp 명령어를 입력하고 password를 입력하면 전송이 완료됩니다.

만약 Are you sure you want to continue connecting (yes/no/[fingerprint])? 문구가 뜨면 yes를 입력해줍시다. 그 다음 password를 칠 수 있습니다.

명령어에 설정한 리눅스 해당 폴더에 가서 확인하시면 됩니다.

반응형

locust를 실행할 때 user 클래스에게 특정 값을 초기에 전달해야하는 경우가 있다.

예를 들면, 특정 도메인으로 요청을 보내고 싶을 때가 있고 아닐 때가 있는데

이 분기를 실행할 때 user 클래스로 True/False를 전달해서 결정할 수 있도록 하는 것이다.

 

 

이번 포스팅은 locust 실행 시 커스텀 인수를 전달하는 2가지 방법에 대해 정리하고자 한다.

 

  1. command line 명령어를 커스터마이징 하기
  2. 환경 변수 설정하기

 


 

 

1. command line 명령어 커스터마이징 하기

 

 

locust에는 디폴트로 설정된 커맨드라인 명령어가 있다. 여기에 내가 커스터마이징 한 명령어를 추가하는 방법을 사용한다

 

이 방법은 locust의 events 데코레이션을 활용하게 된다.

 

 

from locust import events

@events.init_command_line_parser.add_listener	#데코레이터 추가
def set_command_line(parser):
    #명령어 추가
    parser.add_argument("--test", type=bool, default=False, help="let's test customizing")

 

 

코드를 한 줄씩 살펴보면 먼저 events 데코레이션에서 커맨드라인 parser의 add_listener를 사용한다.

 

그리고 parser를 파라미터로 전달받아 argument를 추가한다.

 

add_argument 내부 파라미터는 여러가지가 있는데, 차례로 명령어, 입력타입, 디폴트값, 그리고 설명을 위 코드에 넣었다.

이외의 유용한 파라미터들은 공식문서를 통해 파악하면 된다.

 

위 함수는 user클래스 밖에 선언하면 된다.

 

재밌는 점은, 위와 같이 커맨드라인 명령어를 추가하고 locust -h 를 누르면 디폴트 명령어들과 함께 내가 추가한 명령어까지 콘솔에 나온다는 것이다. help는 이때 출력되는 명령어 설명에 들어가는 부분이다.

 

 

 

위와 같이 명령어를 추가했다면 user 클래스에서 사용해보자.

 

 

class TestUser(HttpUser):
    if self.environment.parsed_options.test == True:	#명령어 입력값 가져오기
    	self.client.get("/test")

 

 

사용은 간단하게 environment에 포함된 parsed_options에서 내가 추가한 명령어의 값을 가져오면 된다.

필자는 --test 라고 작성했으니 사용도 parsed_options.test로 사용하면 되는 것이다.

 

 

 

이제 이 명령어를 콘솔에서 사용해보자

 

 

locust --test True

 

 

 

 

2.  환경 변수 설정하기

 

따로 명령어를 추가하고 사용하는 것이 번거롭다면 단순히 환경변수를 사용하는 방법도 있다.

 

환경변수의 장점은 어떤 코드에서든 입력값에 접근할 수 있다는 점이다.

필자는 TaskSet에서 입력값을 가져와야 했는데 추가한 명령어로 접근하는 경우 user class에서 값을 가져와 TaskSet으로 전달하는 번거로운 과정을 거쳐야 했다. TaskSet은 environment에 접근할 수 없기 때문이다.

 

그러나 환경변수를 사용하면 TaskSet이든 일반 함수든 자유롭게 입력값에 접근할 수 있다.

 

 

환경변수는 따로 추가하거나 설정해줄 필요 없이 콘솔에서 다음과 같이 실행하면 된다.

 

 

<linux/mac>
TEST=True locust ...

<Window>
SET TEST=True
locust ...

 

 

운영체제에 따라 달라지긴 하지만 위처럼 간단하게 환경변수를 추가할 수 있다.

 

이렇게 추가한 변수를 사용해보자

 

 

import os
from locust import TaskSet, task

class TestTaskSet(TaskSet):
	
    @task
    def test_task(self):
    	if os.environ['TEST'] == "True":		# 환경변수 값 가져오기
        	self.client.get("/test")

 

 

사용은 os에서 환경변수를 가져오는 방식과 동일하다.

단, 환경변수는 타입이 str이므로 필요하면 타입을 변경하는 방식으로 사용하면 된다.

반응형

+ Recent posts