Added support for token auth (rebased, ci fixed)

.github/workflows/rust.yml

@@ -39,7 +39,7 @@ jobs:
         runs-on: ${{ matrix.os }}
         strategy:
             matrix:
-                rust_release: ["1.56", stable, nightly]
+                rust_release: ["1.59", stable, nightly]
                 os: [ubuntu-latest, windows-latest, macOS-latest]
 
         steps:
@@ -50,8 +50,24 @@ jobs:
             - name: Build
               run: cargo build --verbose
 
+    unit_test:
+        name: Unit test (${{ matrix.rust_release }}/${{ matrix.os }})
+        runs-on: ${{ matrix.os }}
+        strategy:
+            matrix:
+                rust_release: ["1.59", stable, nightly]
+                os: [ubuntu-latest, windows-latest, macOS-latest]
+
+        steps:
+            - uses: actions/checkout@v1
+            - uses: dtolnay/rust-toolchain@master
+              with:
+                  toolchain: ${{ matrix.rust_release }}
+            - name: test
+              run: cargo test --lib
+
     integration_test:
-        name: Integration Tests (stable/ubuntu-latest)
+        name: Integration Tests for Influxdb 1.x (stable/ubuntu-latest)
         runs-on: ubuntu-latest
         strategy:
             matrix:
@@ -71,11 +87,33 @@ jobs:
                     INFLUXDB_ADMIN_PASSWORD: password
                     INFLUXDB_USER: nopriv_user
                     INFLUXDB_USER_PASSWORD: password
-
         steps:
             - uses: actions/checkout@v3
             - uses: dtolnay/rust-toolchain@stable
-            - run: cargo test --manifest-path=./influxdb/Cargo.toml --no-default-features --features 'use-serde derive ${{ matrix.http-backend }}' --no-fail-fast
+            - run: cargo test --manifest-path=./influxdb/Cargo.toml --no-default-features --features 'use-serde derive ${{ matrix.http-backend }}' --no-fail-fast --test integration_tests
+
+    integration_test_v2:
+        name: Integration Tests for Influxdb 2.0 (stable/ubuntu-latest)
+        runs-on: ubuntu-latest
+        strategy:
+            matrix:
+                http-backend: [curl-client, h1-client, h1-client-rustls, hyper-client]
+        services:
+            influxdbv2:
+                image: influxdb:2.6
+                ports:
+                    - 9086:8086
+                env:
+                    DOCKER_INFLUXDB_INIT_MODE: setup
+                    DOCKER_INFLUXDB_INIT_USERNAME: admin
+                    DOCKER_INFLUXDB_INIT_PASSWORD: password
+                    DOCKER_INFLUXDB_INIT_ORG: testing
+                    DOCKER_INFLUXDB_INIT_BUCKET: mydb
+                    DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: admintoken
+        steps:
+            - uses: actions/checkout@v1
+            - uses: dtolnay/rust-toolchain@stable
+            - run: cargo test --manifest-path=./influxdb/Cargo.toml --no-default-features --features 'use-serde derive ${{ matrix.http-backend }}' --no-fail-fast --test integration_tests_v2
 
     coverage:
         name: Code Coverage (stable/ubuntu-latest)

auxiliary/update_cargo-readme.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/bash
 
 our_version=$(cargo readme -V | perl -ne 'print $1 while  /v([\d.]+)/g')
-last_version=$(cargo search cargo-readme | perl -ne 'print $1 while /^cargo-readme = "([\d.]+)"/g')
+last_version=$(cargo search --color=never cargo-readme | perl -ne 'print $1 while /^cargo-readme = "([\d.]+)"/g')
 
 if [ "$our_version" == "$last_version" ]; then
     echo Version $our_version is of cargo-readme is installed and up to date.

influxdb/src/client/mod.rs

@@ -18,12 +18,12 @@
 use futures_util::TryFutureExt;
 use http::StatusCode;
 #[cfg(feature = "reqwest")]
-use reqwest::{Client as HttpClient, Response as HttpResponse};
+use reqwest::{Client as HttpClient, RequestBuilder, Response as HttpResponse};
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::{self, Debug, Formatter};
 use std::sync::Arc;
 #[cfg(feature = "surf")]
-use surf::{Client as HttpClient, Response as HttpResponse};
+use surf::{Client as HttpClient, RequestBuilder, Response as HttpResponse};
 
 use crate::query::QueryType;
 use crate::Error;
@@ -34,6 +34,7 @@ use crate::Query;
 pub struct Client {
     pub(crate) url: Arc<String>,
     pub(crate) parameters: Arc<HashMap<&'static str, String>>,
+    pub(crate) token: Option<String>,
     pub(crate) client: HttpClient,
 }
 
@@ -89,6 +90,7 @@ impl Client {
             url: Arc::new(url.into()),
             parameters: Arc::new(parameters),
             client: HttpClient::new(),
+            token: None,
         }
     }
 
@@ -126,6 +128,19 @@ impl Client {
         self
     }
 
+    /// Add authorization token to [`Client`](crate::Client)
+    ///
+    /// This is designed for influxdb 2.0's backward-compatible API which
+    /// requires authrozation by default. You can create such token from
+    /// console of influxdb 2.0 .
+    pub fn with_token<S>(mut self, token: S) -> Self
+    where
+        S: Into<String>,
+    {
+        self.token = Some(token.into());
+        self
+    }
+
     /// Returns the name of the database the client is using
     pub fn database_name(&self) -> &str {
         // safe to unwrap: we always set the database name in `Self::new`
@@ -218,11 +233,11 @@ impl Client {
             error: err.to_string(),
         })?;
 
+        let mut parameters = self.parameters.as_ref().clone();
         let request_builder = match q.get_type() {
             QueryType::ReadQuery => {
                 let read_query = query.get();
                 let url = &format!("{}/query", &self.url);
-                let mut parameters = self.parameters.as_ref().clone();
                 parameters.insert("q", read_query.clone());
 
                 if read_query.contains("SELECT") || read_query.contains("SHOW") {
@@ -245,7 +260,8 @@ impl Client {
             error: err.to_string(),
         })?;
 
-        let res = request_builder
+        let res = self
+            .auth_if_needed(request_builder)
             .send()
             .map_err(|err| Error::ConnectionError {
                 error: err.to_string(),
@@ -273,6 +289,14 @@ impl Client {
 
         Ok(s)
     }
+
+    fn auth_if_needed(&self, rb: RequestBuilder) -> RequestBuilder {
+        if let Some(ref token) = self.token {
+            rb.header("Authorization", format!("Token {}", token))
+        } else {
+            rb
+        }
+    }
 }
 
 pub(crate) fn check_status(res: &HttpResponse) -> Result<(), Error> {
@@ -327,5 +351,11 @@ mod tests {
         assert_eq!(with_auth.parameters.get("db").unwrap(), "database");
         assert_eq!(with_auth.parameters.get("u").unwrap(), "username");
         assert_eq!(with_auth.parameters.get("p").unwrap(), "password");
+
+        let client = Client::new("http://localhost:8068", "database");
+        let with_auth = client.with_token("token");
+        assert_eq!(with_auth.parameters.len(), 1);
+        assert_eq!(with_auth.parameters.get("db").unwrap(), "database");
+        assert_eq!(with_auth.token.unwrap(), "token");
     }
 }

influxdb/src/query/line_proto_term.rs

@@ -23,8 +23,8 @@ impl LineProtoTerm<'_> {
     pub fn escape(self) -> String {
         use LineProtoTerm::*;
         match self {
-            Measurement(x) => Self::escape_any(x, &*COMMAS_SPACES),
-            TagKey(x) | FieldKey(x) => Self::escape_any(x, &*COMMAS_SPACES_EQUALS),
+            Measurement(x) => Self::escape_any(x, &COMMAS_SPACES),
+            TagKey(x) | FieldKey(x) => Self::escape_any(x, &COMMAS_SPACES_EQUALS),
             FieldValue(x) => Self::escape_field_value(x),
             TagValue(x) => Self::escape_tag_value(x),
         }
@@ -44,7 +44,7 @@ impl LineProtoTerm<'_> {
             Float(v) => v.to_string(),
             SignedInteger(v) => format!("{}i", v),
             UnsignedInteger(v) => format!("{}u", v),
-            Text(v) => format!(r#""{}""#, Self::escape_any(v, &*QUOTES_SLASHES)),
+            Text(v) => format!(r#""{}""#, Self::escape_any(v, &QUOTES_SLASHES)),
         }
     }
 
@@ -62,7 +62,7 @@ impl LineProtoTerm<'_> {
             Float(v) => format!(r#"{}"#, v),
             SignedInteger(v) => format!(r#"{}"#, v),
             UnsignedInteger(v) => format!(r#"{}"#, v),
-            Text(v) => Self::escape_any(v, &*SLASHES),
+            Text(v) => Self::escape_any(v, &SLASHES),
         }
     }
 

influxdb/src/query/mod.rs

@@ -281,7 +281,11 @@ mod tests {
     }
     #[test]
     fn test_timestamp_from_chrono_date() {
-        let timestamp_from_datetime: Timestamp = Utc.ymd(1970, 1, 1).and_hms(0, 0, 1).into();
+        let timestamp_from_datetime: Timestamp = Utc
+            .with_ymd_and_hms(1970, 1, 1, 0, 0, 1)
+            .single()
+            .unwrap()
+            .into();
         assert_eq!(
             Timestamp::Nanoseconds(MILLIS_PER_SECOND * NANOS_PER_MILLI),
             timestamp_from_datetime

influxdb/tests/derive_integration_tests.rs

@@ -101,6 +101,7 @@ async fn test_write_and_read_option() {
                 .query(&weather_reading.into_query("weather_reading".to_string()))
                 .await;
             assert_result_ok(&write_result);
+
             let query = ReadQuery::new("SELECT time, pressure, wind_strength FROM weather_reading");
             let result = client.json_query(query).await.and_then(|mut db_result| {
                 println!("{:?}", db_result);

influxdb/tests/integration_tests.rs

@@ -215,6 +215,7 @@ async fn test_non_authed_write_and_read() {
 
             let read_query = ReadQuery::new("SELECT * FROM weather");
             let read_result = non_authed_client.query(read_query).await;
+
             assert_result_err(&read_result);
             match read_result {
                 Err(Error::AuthorizationError) => {}

influxdb/tests/integration_tests_v2.rs

@@ -0,0 +1,119 @@
+extern crate influxdb;
+
+#[path = "./utilities.rs"]
+mod utilities;
+use utilities::{assert_result_err, assert_result_ok, run_test};
+
+use influxdb::InfluxDbWriteable;
+use influxdb::{Client, Error, ReadQuery, Timestamp};
+
+/// INTEGRATION TEST
+///
+
+/// This test case tests the Authentication
+#[async_std::test]
+#[cfg(not(tarpaulin))]
+async fn test_authed_write_and_read() {
+    run_test(
+        || async move {
+            let client = Client::new("http://127.0.0.1:9086", "mydb").with_token("admintoken");
+            let write_query = Timestamp::Hours(11)
+                .into_query("weather")
+                .add_field("temperature", 82);
+            let write_result = client.query(&write_query).await;
+            assert_result_ok(&write_result);
+
+            let read_query = ReadQuery::new("SELECT * FROM weather");
+            let read_result = client.query(read_query).await;
+            assert_result_ok(&read_result);
+            assert!(
+                !read_result.unwrap().contains("error"),
+                "Data contained a database error"
+            );
+        },
+        || async move {
+            let client = Client::new("http://127.0.0.1:9086", "mydb").with_token("admintoken");
+            let read_query = ReadQuery::new("DELETE MEASUREMENT weather");
+            let read_result = client.query(read_query).await;
+            assert_result_ok(&read_result);
+            assert!(!read_result.unwrap().contains("error"), "Teardown failed");
+        },
+    )
+    .await;
+}
+
+/// INTEGRATION TEST
+///
+/// This test case tests the Authentication
+#[async_std::test]
+#[cfg(not(tarpaulin))]
+async fn test_wrong_authed_write_and_read() {
+    run_test(
+        || async move {
+            let client = Client::new("http://127.0.0.1:9086", "mydb").with_token("falsetoken");
+            let write_query = Timestamp::Hours(11)
+                .into_query("weather")
+                .add_field("temperature", 82);
+            let write_result = client.query(&write_query).await;
+            assert_result_err(&write_result);
+            match write_result {
+                Err(Error::AuthorizationError) => {}
+                _ => panic!(
+                    "Should be an AuthorizationError: {}",
+                    write_result.unwrap_err()
+                ),
+            }
+
+            let read_query = ReadQuery::new("SELECT * FROM weather");
+            let read_result = client.query(&read_query).await;
+            assert_result_err(&read_result);
+            match read_result {
+                Err(Error::AuthorizationError) => {}
+                _ => panic!(
+                    "Should be an AuthorizationError: {}",
+                    read_result.unwrap_err()
+                ),
+            }
+        },
+        || async move {},
+    )
+    .await;
+}
+
+/// INTEGRATION TEST
+///
+/// This test case tests the Authentication
+#[async_std::test]
+#[cfg(not(tarpaulin))]
+async fn test_non_authed_write_and_read() {
+    run_test(
+        || async move {
+            let non_authed_client = Client::new("http://127.0.0.1:9086", "mydb");
+            let write_query = Timestamp::Hours(11)
+                .into_query("weather")
+                .add_field("temperature", 82);
+            let write_result = non_authed_client.query(&write_query).await;
+            assert_result_err(&write_result);
+            match write_result {
+                Err(Error::AuthorizationError) => {}
+                _ => panic!(
+                    "Should be an AuthorizationError: {}",
+                    write_result.unwrap_err()
+                ),
+            }
+
+            let read_query = ReadQuery::new("SELECT * FROM weather");
+            let read_result = non_authed_client.query(&read_query).await;
+            assert_result_err(&read_result);
+            match read_result {
+                Err(Error::AuthorizationError) => {}
+                _ => panic!(
+                    "Should be an AuthorizationError: {}",
+                    read_result.unwrap_err()
+                ),
+            }
+        },
+        || async move {},
+    )
+    .await;
+}

influxdb/tests/utilities.rs

@@ -14,6 +14,7 @@ pub fn assert_result_ok<A: std::fmt::Debug, B: std::fmt::Debug>(result: &Result<
     result.as_ref().expect("assert_result_ok failed");
 }
 
+#[allow(dead_code)]
 #[cfg(not(tarpaulin_include))]
 pub fn create_client<T>(db_name: T) -> Client
 where
@@ -22,6 +23,7 @@ where
     Client::new("http://127.0.0.1:8086", db_name)
 }
 
+#[allow(dead_code)]
 #[cfg(not(tarpaulin_include))]
 pub async fn create_db<T>(name: T) -> Result<String, Error>
 where
@@ -32,6 +34,7 @@ where
     create_client(test_name).query(ReadQuery::new(query)).await
 }
 
+#[allow(dead_code)]
 #[cfg(not(tarpaulin_include))]
 pub async fn delete_db<T>(name: T) -> Result<String, Error>
 where