client_test.py 文件源码

def test_query(mock_flux):
    db = influxdb.InfluxDBClient(database="fizz")
    db.query.side_effect = influxdb.exceptions.InfluxDBClientError(None)
    client = InfluxAlchemy(db)
    query = client.query(Measurement.new("buzz"))
    assert str(query) == "SELECT * FROM buzz;"