Does aioresponses work for requests in a subprocess?

[liavkoren]

Datadog's api client implements concurrency using subprocesses. If you pass a async_req=True flag in, you get a multiprocessing.pool.ApplyResult instance back. Very quickly testing this with aioresponses doesn't seem to work; is this a pattern that aioresponses is currently set up to handle?

Minimal test case:

reporter.py:

from datetime import datetime

from datadog_api_client.v1 import ApiClient, Configuration
from datadog_api_client.v1.api.metrics_api import MetricsApi
from datadog_api_client.v1.model.metrics_payload import MetricsPayload
from datadog_api_client.v1.model.point import Point
from datadog_api_client.v1.model.series import Series



class DatadogReporter:
    def __init__(self):
        self.config = Configuration()
        self.config.api_key["apiKeyAuth"] = <api-key>
        self.config.api_key["appKeyAuth"] = <app-key>
        self.api_client = ApiClient(self.config)

    def submit_metric(self, body):
        with self.api_client:
            metrics_api = MetricsApi(self.api_client)
            breakpoint()
            try:
                return metrics_api.submit_metrics(body, async_req=True)
            except Exception as e:
                logger.error(f"Failed to submit metric to Datadog: {e}")


def gauge_metric(metric_name, value, tags=None):
    tags = tags or []
    return MetricsPayload(
        series=[
            Series(
                metric=metric_name,
                type="gauge",
                points=[Point([datetime.now().timestamp(), value])],
                tags=tags,
            )
        ]
    )

test_reporter.py:

import aiohttp
import pytest

from app.external.datadog import DatadogReporter, gauge_metric
from app.external.vault import vault


@pytest.fixture
async def aiohttp_session(loop):
    async with aiohttp.ClientSession() as session:
        async_client.session = session
        yield session


@pytest.fixture
def mock_aioresponse(aiohttp_session):
    with aioresponses() as m:
        yield m


class TestDatadogReporter:
    async def test_submit_metric(self, mock_aioresponse):
        reporter = DatadogReporter()
        metric = gauge_metric("test.metric", 1)
        mock_aioresponse.post(
            "https://api.datadoghq.com/api/v1/series",
            status=200,
            payload={"status": "ok foobar"},
        )
        response = reporter.submit_metric(metric)
        assert response.successful()
        assert response.get().to_dict() == {"status": "ok foobar"}

[liavkoren]

Aioresponses is a helper to mock/fake web requests in python aiohttp package.

Ah. As I thought. Gonna close this -- I actually found the datadog python client to be kind of broken and poorly documented, and decided it's easier to just use aiohttp directly.