feat: support-env

Author: NguyenHoangSon96

tests/test_influxdb_client_3.py

@@ -1,7 +1,7 @@
 import unittest
 from unittest.mock import patch
 
-from influxdb_client_3 import InfluxDBClient3
+from influxdb_client_3 import InfluxDBClient3, WritePrecision, DefaultWriteOptions
 from tests.util import asyncio_run
 from tests.util.mocks import ConstantFlightServer, ConstantData
 
@@ -74,6 +74,63 @@ async def test_query_async(self):
             assert {'data': 'sql_query', 'reference': query, 'value': -1.0} in result_list
             assert {'data': 'query_type', 'reference': 'sql', 'value': -1.0} in result_list
 
+    def test_default_client(self):
+        expected_precision = DefaultWriteOptions.write_precision.value
+        expected_write_type = DefaultWriteOptions.write_type.value
+        expected_gzip_enabled = False
+
+        def verify_client_write_options(c):
+            write_options = c._write_client_options.get('write_options')
+            self.assertEqual(write_options.write_precision, expected_precision)
+            self.assertEqual(write_options.write_type, expected_write_type)
+
+            self.assertEqual(c._write_api._write_options.write_precision, expected_precision)
+            self.assertEqual(c._write_api._write_options.write_type, expected_write_type)
+
+        env_client = InfluxDBClient3.from_env()
+        verify_client_write_options(env_client)
+
+        default_client = InfluxDBClient3()
+        verify_client_write_options(default_client)
+
+    @patch.dict('os.environ', {'INFLUX_HOST': 'localhost', 'INFLUX_TOKEN': 'test_token',
+                               'INFLUX_DATABASE': 'test_db', 'INFLUX_ORG': 'test_org',
+                               'INFLUX_PRECISION': WritePrecision.MS, 'INFLUX_AUTH_SCHEME': 'custom_scheme'})
+    def test_from_env_all_env_vars_set(self):
+        client = InfluxDBClient3.from_env()
+        self.assertIsInstance(client, InfluxDBClient3)
+        self.assertEqual(client._token, "test_token")
+        self.assertEqual(client._client.url, "https://localhost:443")
+        self.assertEqual(client._client.auth_header_value, f"custom_scheme {client._token}")
+        self.assertEqual(client._database, "test_db")
+        self.assertEqual(client._org, "test_org")
+
+        write_options = client._write_client_options.get("write_options")
+        self.assertEqual(write_options.write_precision, WritePrecision.MS)
+
+        client._write_api._point_settings = {}
+
+    @patch.dict('os.environ', {'INFLUX_HOST': "", 'INFLUX_TOKEN': "",
+                               'INFLUX_DATABASE': "", 'INFLUX_ORG': ""})
+    def test_from_env_missing_variables(self):
+        with self.assertRaises(ValueError) as context:
+            InfluxDBClient3.from_env()
+        self.assertIn("Missing required environment variables", str(context.exception))
+
+    @patch.dict('os.environ', {'INFLUX_HOST': 'localhost', 'INFLUX_TOKEN': 'test_token',
+                               'INFLUX_DATABASE': 'test_db', 'INFLUX_PRECISION': WritePrecision.MS})
+    def test_parse_valid_write_precision(self):
+        client = InfluxDBClient3.from_env()
+        self.assertIsInstance(client, InfluxDBClient3)
+        self.assertEqual(client._write_client_options.get('write_options').write_precision, WritePrecision.MS)
+
+    @patch.dict('os.environ', {'INFLUX_HOST': 'localhost', 'INFLUX_TOKEN': 'test_token',
+                               'INFLUX_DATABASE': 'test_db', 'INFLUX_PRECISION': 'invalid_value'})
+    def test_parse_invalid_write_precision(self):
+        with self.assertRaises(ValueError) as context:
+            InfluxDBClient3.from_env()
+        self.assertIn("Invalid precision value: invalid_value", str(context.exception))
+
 
 if __name__ == '__main__':
     unittest.main()