mardi 18 juillet 2017

Using pytest to mock multiple queries to a Stackdriver API

I am using pytest to test some code that queries the StackDriver monitoring API.

Basically the actual code loops over a list of metrics and then queries the api with each metric, assigning it to a dict. It then calculates an error rate by via the results in the metric dictionary.

First I iterate over my metric list, querying the stackdriver client in each iteration:

    metric_list = ['a', 'b']

    for metric in metric_list:
        query = sd.query(
            metric, end_time=end_time, minutes=1
        )

I then iterate over the query, getting a list of the point values and assign them to a special dict:

        for time_series in query:
            metric_dict[metric] = sum([point.value for point in time_series.points])

time_series.points basically returns a list of namedtuples that look like so:

Points(
   end_time='2017-07-14T14:26:06750Z',
   start_time='2017-07-14T14:25:06.750Z',
   value=1), Points(
   end_time='2017-07-14T14:27:06.750Z',
   start_time='2017-07-14T1426:06.750Z',
   value=1)

The dict would just look like a list of numbers assigned to each metric key.

I then need to sum the entire dict of values like so:

    total = sum(metric_dict.values()) 

After which I need to grab a specific metric from the metric_list, and then use it as a key so I can work out the error rate based on the metric types:

    ok_metric = [metric for metric in metric_list if 'ok' in metric]

    return ( 
            100 - abs(((metric_dict[ok_metric[0]] - total) / (metric_dict[ok_metric[0]])) * 100)
    )

So I am able to mock the return_value of one query object, but where I am struggling is mocking how the script iterates over the metric_list and creates more than one return_value.

Can anyone suggest the best way of mocking this?

At the moment I have this:

TimeSeries = namedtuple('TimeSeries', 'points')
Points = namedtuple('Points', 'end_time start_time value')
Value = namedtuple('Value', 'value')

error_points =  Points(
                end_time='2017-07-14T14:26:06750Z', 
                start_time='2017-07-14T14:25:06.750Z', 
                value=1), Points(
                end_time='2017-07-14T14:27:06.750Z', 
                start_time='2017-07-14T1426:06.750Z', 
                value=1)

ok_points =  Points(
                end_time='2017-07-14T14:26:06750Z', 
                start_time='2017-07-14T14:25:06.750Z', 
                value=1), Points(
                end_time='2017-07-14T14:27:06.750Z', 
                start_time='2017-07-14T1426:06.750Z', 
                value=1)

time_series_ok = TimeSeries(ok_points)                
time_series_error = TimeSeries(error_points)
mock_data = [time_series_ok, time_series_error]

mock_sd_metric = mocker.patch.object(sources.sd, 'query')

for data in mock_data:
    mock_sd_metric.return_value = [data]

But it's not working for (probably) obvious reasons.

Thanks in advance.

Aucun commentaire:

Enregistrer un commentaire