test_influxdb_integration.py 文件源码

def test_multi_fetch_non_existant_series(self):
        """Test single fetch data for a series by name"""
        path1, path2 = 'fake_path1', 'fake_path2'
        reader1 = influxgraph.InfluxDBReader(InfluxDBClient(
            database=self.db_name), path1)
        reader2 = influxgraph.InfluxDBReader(InfluxDBClient(
            database=self.db_name), path2)
        nodes = [influxgraph.classes.leaf.InfluxDBLeafNode(path1, reader1),
                 influxgraph.classes.leaf.InfluxDBLeafNode(path2, reader2)]
        time_info, data = self.finder.fetch_multi(nodes,
                                                  int(self.start_time.strftime("%s")),
                                                  int(self.end_time.strftime("%s")))
        for metric_name in data:
            self.assertFalse(data[metric_name],
                             msg="Expected no data for non-existant series %s - got %s" % (
                                 metric_name, data,))
        fake_nodes = list(self.finder.find_nodes(Query('fake_pathy_path')))
        time_info, data = self.finder.fetch_multi(fake_nodes,
                                                  int(self.start_time.strftime("%s")),
                                                  int(self.end_time.strftime("%s")))
        self.assertFalse(data)