Тестирование интеграции Определение исходного оператора в автономном режиме с помощью фикстур ответов

Внедрите транспортный уровень и протестируйте явные синтетические фикстуры ответов, прежде чем разрешать потребителю CarrierLookup использовать сеть.

Модульные тесты API для определения оператора должны заменять HTTP-транспорт, блокировать случайный доступ к сети и проверять, как ваше приложение обрабатывает каждый задокументированный результат. Синтетические фикстуры позволяют тестировать логику потребителя без отправки телефонных номеров или расходования баланса.

Разместите точку тестирования перед HTTP-запросом

Импортер CRM может успешно пройти тест «счастливого пути» и при этом пометить номер со статусом Не определено как недействительный. Он также может пройти тест в эмулируемом браузере, в то время как его бэкенд выполняет реальный платный запрос. Разместите точку тестирования в той функции бэкенда, которая выполняет исходящий HTTP-обмен.

Потребитель должен принимать транспортную зависимость, а не обнаруживать реальный клиент внутри себя. Внедренный тестовый транспорт получает целевой путь и тело запроса, проверяет их и возвращает синтетический конверт. В тесте нет API-ключа. Также не храните рабочие учетные данные в задании CI.

Используйте отдельное ограничение сети в качестве резервного механизма. Фикстура, возвращающая удобные данные, не является доказательством того, что все исходящие вызовы были перехвачены. Запретите исходящий трафик на уровне контейнера CI или исполнителя и завершайте задание с ошибкой при попытке доступа к сети. Защита на уровне языка помогает выявить распространенные ошибки, но не может охватить подпроцессы или все сетевые библиотеки.

Определите результаты перед написанием фикстур

Наш синхронный сервис оператора использует service_type=carrier. Одиночные ответы имеют плоские поля data, включая identifier, carrier, underlying_carrier, number_type, country_code, region и city. Чтение data.extra.carrier проверяет вымышленную структуру.

Значения оператора описывают исходное распределение. Они не должны заполнять поле CRM, называемое текущей сетью. Аналогично, ни пустой результат распределения, ни exists=false не подтверждают действительность номера. Используйте отдельные метки результатов приложения, чтобы последующий отчетный запрос не мог случайно объединить эти состояния.

Запустите небольшой набор синтетических фикстур

Сохраните следующий код как carrier_consumer_test.py и выполните python3 carrier_consumer_test.py. Он использует стандартную библиотеку Python. Пример представляет собой минимальный потребительский интерфейс с синтетическими идентификаторами и намеренно выбранными значениями; он никогда не связывается с CarrierLookup. Замените демонстрационный client и классификатор своим собственным адаптером, сохраняя при этом утверждения фикстур.

import socket
import unittest
from unittest.mock import patch

FIELDS = ("carrier", "underlying_carrier", "number_type",
          "country_code", "region", "city")
A, B = "+12025550101", "+12025550102"  # Только синтетические тестовые данные.

def classify(envelope):
    code = envelope["code"]
    if code != 0:
        return {42200: "undetermined", 42901: "busy",
                50400: "timeout"}.get(code, "error")
    data = envelope["data"]
    if data["service_type"] != "carrier":
        raise ValueError("unexpected service")
    return "completed" if data["carrier"] else "completed_empty"

def client(transport, identifiers):
    multi = len(identifiers) > 1
    path = "/api/v1/batch-check" if multi else "/api/v1/check"
    body = {"service_type": "carrier"}
    body["identifiers" if multi else "identifier"] = (
        identifiers if multi else identifiers[0])
    return transport(path, body)

def completed(identifier, carrier=""):
    return {"code": 0, "msg": "synthetic fixture", "data": {
        "service_type": "carrier", "identifier": identifier,
        **dict.fromkeys(FIELDS, ""), "carrier": carrier}}

class ConsumerTests(unittest.TestCase):
    def setUp(self):
        self.guard = patch.object(
            socket.socket, "connect",
            side_effect=AssertionError("network forbidden"))
        self.guard.start()
        self.addCleanup(self.guard.stop)

    def test_single_flat_data_and_outcomes(self):
        for value, expected in [("Example Allocation", "completed"),
                                ("", "completed_empty")]:
            def fake(path, body):
                self.assertEqual(path, "/api/v1/check")
                self.assertEqual(body, {"service_type": "carrier",
                                        "identifier": A})
                return completed(A, value)
            result = client(fake, [A])
            self.assertNotIn("extra", result["data"])
            self.assertNotIn("registered", result["data"])
            self.assertEqual(classify(result), expected)

    def test_error_branches(self):
        for code, expected in [(42200, "undetermined"),
                               (42901, "busy"), (50400, "timeout")]:
            with self.subTest(code=code):
                result = client(lambda path, body: {
                    "code": code, "msg": "synthetic error",
                    "data": None}, [A])
                self.assertEqual(classify(result), expected)

    def test_multi_order_and_no_result(self):
        def fake(path, body):
            self.assertEqual(path, "/api/v1/batch-check")
            self.assertEqual(body, {"service_type": "carrier",
                                    "identifiers": [B, A]})
            return {"code": 0, "msg": "synthetic fixture", "data": {
                "service_type": "carrier", "total": 2,
                "succeeded": 1, "failed": 1, "results": [
                    {"identifier": B, "exists": True,
                     **dict.fromkeys(FIELDS, "")},
                    {"identifier": A, "exists": False}]}}
        data = client(fake, [B, A])["data"]
        rows = data["results"]
        self.assertEqual(data["service_type"], "carrier")
        self.assertEqual([r["identifier"] for r in rows], [B, A])
        self.assertEqual(len(rows), data["total"])
        outcomes = ["completed" if r["exists"] else "no_result"
                    for r in rows]
        self.assertEqual(outcomes, ["completed", "no_result"])
        self.assertNotIn("carrier", rows[1])

    def test_guard_rejects_an_escape(self):
        with socket.socket() as sock:
            with self.assertRaisesRegex(AssertionError, "network forbidden"):
                sock.connect(("127.0.0.1", 9))

if __name__ == "__main__":
    unittest.main()

Пример возвращает уже декодированные конверты из своего транспорта. Он тестирует построение запроса и обработку результатов, а не HTTP-парсинг. Сохраняйте фикстуры для HTTP-статусов, заголовков и некорректного JSON также на границе вашего реального транспорта. В частности, ответ 42901 включает Retry-After; фикстура ответа должна проверять, что ваш транспорт сохраняет его для вызывающей стороны. Не встраивайте жестко новый интервал повтора в классификатор результатов.

Защита сокета намеренно тестирует свой собственный путь отказа. Она блокирует метод подключения сокета этого процесса Python; отдельное ограничение исходящего трафика CI остается необходимым для других путей. Случайный платный вызов должен приводить к сбою задания, а не становиться незамеченной дополнительной фикстурой.

Сделайте утверждения для нескольких результатов чувствительными к ошибкам порядка

Фикстура для нескольких элементов отправляет B перед A. Это важно: тест, который отправляет отсортированные по алфавиту идентификаторы и получает такой же отсортированный порядок, может пропустить адаптер, который сортирует ответы. Сопоставляйте каждый возвращенный идентификатор с его позицией ввода и проверяйте общую длину.

Первый элемент в запросе нескольких номеров создает завершенный результат с пустыми значениями распределения. Второй имеет exists=false и не содержит полей оператора. Сохраняйте это различие при сопоставлении с записями приложения. Если ваш производственный адаптер отбрасывает неудачные строки, фикстура должна завершиться ошибкой, так как в противном случае последующие строки могут сместиться на неверные контакты.

Добавьте одно утверждение в фактическую функцию сопоставления CRM: неопределенный или неудачный поиск должен оставлять предыдущее наблюдение распределения нетронутым или явно помечать его как устаревшее в соответствии с вашей политикой. Он не должен перезаписывать запись как «недействительный номер». Тот же тест должен отклонять попытки вывода текущего местоположения из региона или города распределения.

Синхронный тайм-аут для нескольких номеров приводит к сбою всего запроса без частичных результатов. Тестируйте ветвь тайм-аута всего запроса отдельно от успешного конверта, содержащего неудачный элемент. Это разные входные данные для потребителя.

Разделяйте автономную уверенность и проверку в реальном времени

Эти тесты устанавливают, как ваш код реагирует на контракт, представленный фикстурами. Они не демонстрируют текущую доступность провайдера, принятие страны, разрешения учетной записи или биллинг. Созданная фикстура также не может доказать, что реальный провайдер вернет определенное поле распределения для конкретного номера.

Храните происхождение фикстуры рядом с тестом: CarrierLookup API reference, дату проверки и поля, которые использует ваш адаптер. Обновляйте фикстуры при изменении задокументированного контракта, а затем проверяйте, нужны ли старым данным приложения совместимые пути чтения.

Запускайте отдельно авторизованный «дымовой» тест в реальном времени, когда этого требует проверка интеграции, используя утвержденные входные данные и собственный бюджет баланса. Не делайте обычный CI зависимым от этого вызова. Начните с замены одного импорта реального клиента в вашем наборе тестов на внедренный транспорт и подтверждения того, что попытка сетевого выхода немедленно завершается ошибкой.

Узнать больше

Выберите информацию о продукте, которая соответствует следующему шагу в вашем рабочем процессе.

• Просмотреть документацию по продукту →
• Связаться с командой →