반응형
Uvicorn + Fast를 종료하는 방법pytest로 테스트 시 workers > = 2로 API를 깔끔하게 적용
나는 Uvicorn + Fast로 작성된 애플리케이션이 있다.API. PyTest를 이용하여 응답 시간을 테스트하고 있습니다.
를 참고하여 테스트를 작성했습니다. 하지만 작업자 >= 2번일 때 테스트를 완료한 후 지원 과정이 살아있는 것을 발견했습니다.
나는 시험이 끝나면 지원 절차를 깨끗이 종료하고 싶다.
당신은 알기라도 하나요?
자세한 내용은 다음과 같습니다.
환경
- 윈도우 10
- 바시 4.4.23()
- 파이썬 3.7.5
라이브러리
- fastapi == 0.68.0
- 유비콘 == 0.14.0
- 요청 == 2.26.0
- 파이 검정 == 6.2.4
샘플 코드
- 용도: 메인.파이의
- 테스트: test_main.파이의
실행 결과
$ curl http://localhost:8765
curl: (7) Failed to connect to localhost port 8765: Connection refused
$ pytest test_main.py
=============== test session starts =============== platform win32 -- Python 3.7.5, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: .\
collected 1 item
test_main.py . [100%]
=============== 1 passed in 20.23s ===============
$ curl http://localhost:8765
curl: (7) Failed to connect to localhost port 8765: Connection refused
$ sed -i -e "s/WORKERS = 1/WORKERS = 3/g" test_main.py
$ curl http://localhost:8765
curl: (7) Failed to connect to localhost port 8765: Connection refused
$ pytest test_main.py
=============== test session starts =============== platform win32 -- Python 3.7.5, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: .\
collected 1 item
test_main.py . [100%]
=============== 1 passed in 20.21s ===============
$ curl http://localhost:8765
"hello world"
$ # Why is localhost:8765 still alive?
나는 스스로 해결책을 찾았다.
감사합니다 >
해결책
psutil을 설치한 후 test_main을 업데이트합니다.파이의
from multiprocessing import Process
import psutil
import pytest
import requests
import time
import uvicorn
HOST = "127.0.0.1"
PORT = 8765
WORKERS = 3
def run_server(host: str, port: int, workers: int, wait: int = 15) -> Process:
proc = Process(
target=uvicorn.run,
args=("main:app",),
kwargs={
"host": host,
"port": port,
"workers": workers,
},
)
proc.start()
time.sleep(wait)
assert proc.is_alive()
return proc
def shutdown_server(proc: Process):
##### SOLUTION #####
pid = proc.pid
parent = psutil.Process(pid)
for child in parent.children(recursive=True):
child.kill()
##### SOLUTION END ####
proc.terminate()
for _ in range(5):
if proc.is_alive():
time.sleep(5)
else:
return
else:
raise Exception("Process still alive")
def check_response(host: str, port: int):
assert requests.get(f"http://{host}:{port}").text == '"hello world"'
def check_response_time(host: str, port: int, tol: float = 1e-2):
s = time.time()
requests.get(f"http://{host}:{port}")
e = time.time()
assert e-s < tol
@pytest.fixture(scope="session")
def server():
proc = run_server(HOST, PORT, WORKERS)
try:
yield
finally:
shutdown_server(proc)
def test_main(server):
check_response(HOST, PORT)
check_response_time(HOST, PORT)
check_response(HOST, PORT)
check_response_time(HOST, PORT)
실행 결과
$ curl http://localhost:8765
curl: (7) Failed to connect to localhost port 8765: Connection refused
$ pytest test_main.py
================== test session starts ================== platform win32 -- Python 3.7.5, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: .\
collected 1 item
test_main.py . [100%]
================== 1 passed in 20.24s ==================
$ curl http://localhost:8765
curl: (7) Failed to connect to localhost port 8765: Connection refused
답변 확인
import os
import fastapi
import uvicorn
import psutil
@app.get("/quit")
def iquit():
parent_pid = os.getpid()
parent = psutil.Process(parent_pid)
for child in parent.children(recursive=True): # or parent.children() for recursive=False
child.kill()
parent.kill()
if __name__ == '__main__':
uvicorn.run(app, port=36113, host='127.0.0.1')
반응형
'개발하자' 카테고리의 다른 글
Flot : ListView에 헤더 행을 추가하는 방법 (0) | 2022.12.13 |
---|---|
테라폼 플랫 튜플과 문자열 집합 (0) | 2022.12.12 |
Flot Firebase Auth Exception의 오류 코드는 무엇입니까? (0) | 2022.12.11 |
사이드 메뉴바를 펄럭이는 방법은? (0) | 2022.12.10 |
Azure CNI 네트워크 정책을 사용하여 Kubernetes 포드에서 특정 FQDN/DNS로의 송신만 허용 (0) | 2022.12.10 |