(
        self,
        name,
        value=None,
        tags=None,
        count=None,
        at_least=1,
        hostname=None,
        metric_type=None,
        device=None,
        flush_first_value=None,
    )
Assert a metric was processed by this stub
Assert a metric was processed by this stub
Assert a metric was processed by this stub
def assert_metric(
        self,
        name,
        value=None,
        tags=None,
        count=None,
        at_least=1,
        hostname=None,
        metric_type=None,
        device=None,
        flush_first_value=None,
    ):
        """
        Assert a metric was processed by this stub
        """

        self._asserted.add(name)
        expected_tags = normalize_tags(tags, sort=True)

        candidates = []
        for metric in self.metrics(name):
            if value is not None and not self.is_aggregate(metric.type) and value != metric.value:
                continue

            if expected_tags and expected_tags != sorted(metric.tags):
                continue

            if hostname is not None and hostname != metric.hostname:
                continue

            if metric_type is not None and metric_type != metric.type:
                continue

            if device is not None and device != metric.device:
                continue

            if flush_first_value is not None and flush_first_value != metric.flush_first_value:
                continue

            candidates.append(metric)

        expected_metric = MetricStub(name, metric_type, value, tags, hostname, device, flush_first_value)

        if value is not None and candidates and all(self.is_aggregate(m.type) for m in candidates):
            got = sum(m.value for m in candidates)
            msg = "Expected count value for '{}': {}, got {}".format(name, value, got)
            condition = value == got
        elif count is not None:
            msg = "Needed exactly {} candidates for '{}', got {}".format(count, name, len(candidates))
            condition = len(candidates) == count
        else:
            msg = "Needed at least {} candidates for '{}', got {}".format(at_least, name, len(candidates))
            condition = len(candidates) >= at_least
        self._assert(condition, msg=msg, expected_stub=expected_metric, submitted_elements=self._metrics)
def assert_metric ( self , name , value = None , tags = None , count = None , at_least = 1 , hostname = None , metric_type = None , device = None , flush_first_value = None , ) : self . _asserted . add ( name ) expected_tags = normalize_tags ( tags , sort = True ) candidates = [ ] for metric in self . metrics ( name ) : if value is not None and not self . is_aggregate ( metric . type ) and value != metric . value : continue if expected_tags and expected_tags != sorted ( metric . tags ) : continue if hostname is not None and hostname != metric . hostname : continue if metric_type is not None and metric_type != metric . type : continue if device is not None and device != metric . device : continue if flush_first_value is not None and flush_first_value != metric . flush_first_value : continue candidates . append ( metric ) expected_metric = MetricStub ( name , metric_type , value , tags , hostname , device , flush_first_value ) if value is not None and candidates and all ( self . is_aggregate ( m . type ) for m in candidates ) : got = sum ( m . value for m in candidates ) msg = "Expected count value for '{}': {}, got {}" . format ( name , value , got ) condition = value == got elif count is not None : msg = "Needed exactly {} candidates for '{}', got {}" . format ( count , name , len ( candidates ) ) condition = len ( candidates ) == count else : msg = "Needed at least {} candidates for '{}', got {}" . format ( at_least , name , len ( candidates ) ) condition = len ( candidates ) >= at_least self . _assert ( condition , msg = msg , expected_stub = expected_metric , submitted_elements = self . _metrics )