bugbug/http_service/tests/test_integration.py

78 строки
2.2 KiB
Python

# -*- coding: utf-8 -*-
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
import time
from logging import INFO, basicConfig, getLogger
import requests
basicConfig(level=INFO)
logger = getLogger(__name__)
BUGBUG_HTTP_SERVER = os.environ.get("BUGBUG_HTTP_SERVER", "http://localhost:8000/")
# Test classifying a single bug.
def integration_test_single():
timeout = 1200
for _ in range(timeout):
response = requests.get(
f"{BUGBUG_HTTP_SERVER}/defectenhancementtask/predict/1376406",
headers={"X-Api-Key": "integration_test_single"},
)
if response.status_code == 200:
break
time.sleep(1)
response_json = response.json()
if not response.ok:
raise requests.HTTPError(
f"Couldn't get an answer in {timeout} seconds: {response_json}",
response=response,
)
logger.info("Response for bug 1376406 %s", response_json)
assert response_json["class"] is not None
# Test classifying a batch of bugs.
def integration_test_batch():
timeout = 100
for _ in range(timeout):
response = requests.post(
f"{BUGBUG_HTTP_SERVER}/defectenhancementtask/predict/batch",
headers={"X-Api-Key": "integration_test_batch"},
json={"bugs": [1376544, 1376412]},
)
if response.status_code == 200:
break
time.sleep(1)
response_json = response.json()
if not response.ok:
raise requests.HTTPError(
f"Couldn't get an answer in {timeout} seconds: {response_json}",
response=response,
)
response_1376544 = response_json["bugs"]["1376544"]
logger.info("Response for bug 1376544 %s", response_1376544)
assert response_1376544["class"] is not None
response_1376412 = response_json["bugs"]["1376412"]
logger.info("Response for bug 1376412 %s", response_1376412)
assert response_1376412["class"] is not None
if __name__ == "__main__":
integration_test_single()
integration_test_batch()