Merge ff3c3a7 into 161dfd5

Author: ZPascal

docs/content/grafana_api/model.md

@@ -90,6 +90,7 @@ The class includes all necessary variables to establish a connection to the Graf
 - `ssl_context` _ssl.SSLContext_ - Specify the custom ssl context of the Grafana system
 - `num_pools` _int_ - Specify the number of the connection pool
 - `retries` _any_ - Specify the number of the retries. Please use False as parameter to disable the retries
+- `follow_redirects` _bool_ - Specify if redirections should be followed (default True)
 
 <a id="model.DatasourceQuery"></a>
 
@@ -421,12 +422,12 @@ The class includes all necessary variables to generate an Alertmanager config ob
 
 **Arguments**:
 
-- `global_config` _dict_ - Specify the global config of the Alertmanager
-- `inhibit_rules` _list_ - Specify the inhibit rules of the Alertmanager
-- `mute_time_intervals` _list_ - Specify the mute time intervals of the Alertmanager
+- `global_config` _Union[dict, None]_ - Specify the global config of the Alertmanager
+- `inhibit_rules` _Union[list, None]_ - Specify the inhibit rules of the Alertmanager
+- `mute_time_intervals` _Union[list, None]_ - Specify the mute time intervals of the Alertmanager
 - `receivers` _list_ - Specify the receiver's of the Alertmanager
 - `route` _dict_ - Specify the route of the Alertmanager
-- `templates` _list_ - Specify an Alertmanager template
+- `templates` _Union[list, None]_ - Specify an Alertmanager template
 
 <a id="model.AlertmanagerReceivers"></a>
 
@@ -442,15 +443,16 @@ The class includes all necessary variables to generate an Alertmanager receiver'
 **Arguments**:
 
 - `name` _str_ - Specify the name of the receiver
-- `email_configs` _list_ - Specify the email configuration of the receiver's
+- `email_configs` _Union[list, None]_ - Specify the email configuration of the receiver's
 - `grafana_managed_receiver_configs` _list_ - Specify the Grafana managed receiver configuration of the receiver's
-- `opsgenie_configs` _list_ - Specify the ops genie configuration of the receiver's
-- `pagerduty_configs` _dict_ - Specify the pager duty configuration of the receiver's
-- `pushover_configs` _list_ - Specify the push over configuration of the receiver's
-- `slack_configs` _list_ - Specify the Slack configuration of the receiver's
-- `victorops_configs` _list_ - Specify the victor ops configuration of the receiver's
-- `webhook_configs` _list_ - Specify the webhook configuration of the receiver's
-- `wechat_configs` _list_ - Specify the wechaty configuration of the receiver's
+- `opsgenie_configs` _Union[list, None]_ - Specify the ops genie configuration of the receiver's
+- `pagerduty_configs` _Union[list, None]_ - Specify the pager duty configuration of the receiver's
+- `pushover_configs` _Union[list, None]_ - Specify the push over configuration of the receiver's
+- `slack_configs` _Union[list, None]_ - Specify the Slack configuration of the receiver's
+- `sns_configs` _Union[list, None]_ - Specify the SNS configuration of the receiver's
+- `victorops_configs` _Union[list, None]_ - Specify the victor ops configuration of the receiver's
+- `webhook_configs` _Union[list, None]_ - Specify the webhook configuration of the receiver's
+- `wechat_configs` _Union[list, None]_ - Specify the wechaty configuration of the receiver's
 
 <a id="model.RulerRule"></a>
 

grafana_api/alerting.py

@@ -773,11 +773,11 @@ def get_ruler_group(
             )
             raise ValueError
 
-    def test_rule(self, data_query: list) -> dict:
+    def test_rule(self, data_queries: list) -> dict:
         """The method includes a functionality to test a rule specified by a list of datasource rule query objects
 
         Args:
-            data_query (list): Specify a list of datasource rule query objects
+            data_queries (list): Specify a list of datasource rule query objects
 
         Raises:
             ValueError: Missed specifying a necessary value
@@ -787,11 +787,11 @@ def test_rule(self, data_query: list) -> dict:
             api_call (str): Returns the result of the specified query
         """
 
-        if data_query != list():
+        if data_queries != list():
             datasource_rule_query_objects_json: list = list()
             datasource_rule_query_object_json: dict = dict()
 
-            for datasource_rule_query_object in data_query:
+            for datasource_rule_query_object in data_queries:
                 datasource_rule_query_object_json["datasourceUid"] = (
                     datasource_rule_query_object.datasource_uid
                 )
@@ -828,22 +828,22 @@ def test_rule(self, data_query: list) -> dict:
             else:
                 return api_call
         else:
-            logging.error("There is no data_query defined.")
+            logging.error("There is no data_queries defined.")
             raise ValueError
 
     def test_datasource_uid_rule(
         self,
         expr: str,
         condition: str,
-        data_query: list,
+        data_queries: list,
         datasource_uid: str = "grafana",
     ) -> dict:
         """The method includes a functionality to test a datasource uid rule specified by the expr, the condition, a list of data queries and the datasource_uid
 
         Args:
             expr (str): Specify a list of datasource rule query objects
             condition (str): Specify the condition
-            data_query (list): Specify a list of datasource rule query objects
+            data_queries (list): Specify a list of datasource rule query objects
             datasource_uid (str): Specify the datasource uid or recipient of the alerts (default grafana)
 
         Raises:
@@ -858,12 +858,12 @@ def test_datasource_uid_rule(
             len(datasource_uid) != 0
             and len(expr) != 0
             and len(condition) != 0
-            and data_query != list()
+            and data_queries != list()
         ):
             datasource_rule_query_objects_json: list = list()
             datasource_rule_query_object_json: dict = dict()
 
-            for datasource_rule_query_object in data_query:
+            for datasource_rule_query_object in data_queries:
                 datasource_rule_query_object_json["datasourceUid"] = (
                     datasource_rule_query_object.datasource_uid
                 )
@@ -889,7 +889,7 @@ def test_datasource_uid_rule(
                 json.dumps(
                     {
                         "expr": expr,
-                        "grafana_alert": {
+                        "grafana_condition": {
                             "condition": condition,
                             "data": datasource_rule_query_objects_json,
                             "now": str(datetime.datetime.now()),
@@ -905,16 +905,16 @@ def test_datasource_uid_rule(
                 return api_call
         else:
             logging.error(
-                "There is no datasource_uid, expr, condition or data_query defined."
+                "There is no datasource_uid, expr, condition or data_queries defined."
             )
             raise ValueError
 
-    def test_backtest_rule(self, condition: str, data_query: list) -> dict:
-        """The method includes a functionality to test a rule specified by the condition, a list of data queries
+    def test_backtest_rule(self, condition: str, data_queries: list) -> dict:
+        """The method includes a functionality to test a rule specified by the condition and a list of data queries
 
         Args:
             condition (str): Specify the condition
-            data_query (list): Specify a list of datasource rule query objects
+            data_queries (list): Specify a list of datasource rule query objects
 
         Raises:
             ValueError: Missed specifying a necessary value
@@ -924,11 +924,11 @@ def test_backtest_rule(self, condition: str, data_query: list) -> dict:
             api_call (dict): Returns the result of the specified rule
         """
 
-        if len(condition) != 0 and data_query != list():
+        if len(condition) != 0 and data_queries != list():
             datasource_rule_query_objects_json: list = list()
             datasource_rule_query_object_json: dict = dict()
 
-            for datasource_rule_query_object in data_query:
+            for datasource_rule_query_object in data_queries:
                 datasource_rule_query_object_json["datasourceUid"] = (
                     datasource_rule_query_object.datasource_uid
                 )
@@ -966,7 +966,7 @@ def test_backtest_rule(self, condition: str, data_query: list) -> dict:
                 return api_call
         else:
             logging.error(
-                "There is no datasource_uid, condition or data_query defined."
+                "There is no condition or data_queries defined."
             )
             raise ValueError
 

grafana_api/api.py

@@ -343,6 +343,7 @@ def create_the_http_api_client(
                 headers=headers,
                 transport=async_transport,
                 verify=self.grafana_api_model.ssl_context,
+                follow_redirects=self.grafana_api_model.follow_redirects,
             )
         else:
             return httpx.Client(
@@ -351,4 +352,5 @@ def create_the_http_api_client(
                 headers=headers,
                 transport=transport,
                 verify=self.grafana_api_model.ssl_context,
+                follow_redirects=self.grafana_api_model.follow_redirects,
             )

grafana_api/model.py

@@ -86,6 +86,7 @@ class APIModel:
         ssl_context (ssl.SSLContext): Specify the custom ssl context of the Grafana system
         num_pools (int): Specify the number of the connection pool
         retries (any): Specify the number of the retries. Please use False as parameter to disable the retries
+        follow_redirects (bool): Specify if redirections should be followed (default True)
     """
 
     host: str
@@ -98,6 +99,7 @@ class APIModel:
     ssl_context: ssl.SSLContext = httpx.create_ssl_context()
     num_pools: int = 10
     retries: any = 10
+    follow_redirects: bool = True
 
 
 @dataclass
@@ -438,20 +440,20 @@ class AlertmanagerConfig:
     """The class includes all necessary variables to generate an Alertmanager config object that is necessary to communicate and set up the Grafana Alertmanager endpoint
 
     Args:
-        global_config (dict): Specify the global config of the Alertmanager
-        inhibit_rules (list): Specify the inhibit rules of the Alertmanager
-        mute_time_intervals (list): Specify the mute time intervals of the Alertmanager
+        global_config (Union[dict, None]): Specify the global config of the Alertmanager
+        inhibit_rules (Union[list, None]): Specify the inhibit rules of the Alertmanager
+        mute_time_intervals (Union[list, None]): Specify the mute time intervals of the Alertmanager
         receivers (list):  Specify the receiver's of the Alertmanager
         route (dict): Specify the route of the Alertmanager
-        templates (list): Specify an Alertmanager template
+        templates (Union[list, None]): Specify an Alertmanager template
     """
 
-    global_config: dict
-    inhibit_rules: list
-    mute_time_intervals: list
+    global_config: Union[dict, None]
+    inhibit_rules: Union[list, None]
+    mute_time_intervals: Union[list, None]
     receivers: list
     route: dict
-    templates: list
+    templates: Union[list, None]
 
 
 @dataclass
@@ -460,28 +462,29 @@ class AlertmanagerReceivers:
 
     Args:
         name (str): Specify the name of the receiver
-        email_configs (list): Specify the email configuration of the receiver's
+        email_configs (Union[list, None]): Specify the email configuration of the receiver's
         grafana_managed_receiver_configs (list): Specify the Grafana managed receiver configuration of the receiver's
-        opsgenie_configs (list):  Specify the ops genie configuration of the receiver's
-        pagerduty_configs (dict): Specify the pager duty configuration of the receiver's
-        pushover_configs (list): Specify the push over configuration of the receiver's
-        slack_configs (list): Specify the Slack configuration of the receiver's
-        victorops_configs (list): Specify the victor ops configuration of the receiver's
-        webhook_configs (list): Specify the webhook configuration of the receiver's
-        wechat_configs (list): Specify the wechaty configuration of the receiver's
+        opsgenie_configs (Union[list, None]):  Specify the ops genie configuration of the receiver's
+        pagerduty_configs (Union[list, None]): Specify the pager duty configuration of the receiver's
+        pushover_configs (Union[list, None]): Specify the push over configuration of the receiver's
+        slack_configs (Union[list, None]): Specify the Slack configuration of the receiver's
+        sns_configs (Union[list, None]): Specify the SNS configuration of the receiver's
+        victorops_configs (Union[list, None]): Specify the victor ops configuration of the receiver's
+        webhook_configs (Union[list, None]): Specify the webhook configuration of the receiver's
+        wechat_configs (Union[list, None]): Specify the wechaty configuration of the receiver's
     """
 
     name: str
-    email_configs: list
+    email_configs: Union[list, None]
     grafana_managed_receiver_configs: list
-    opsgenie_configs: list
-    pagerduty_configs: list
-    pushover_configs: list
-    slack_configs: list
-    sns_configs: list
-    victorops_configs: list
-    webhook_configs: list
-    wechat_configs: list
+    opsgenie_configs: Union[list, None]
+    pagerduty_configs: Union[list, None]
+    pushover_configs: Union[list, None]
+    slack_configs: Union[list, None]
+    sns_configs: Union[list, None]
+    victorops_configs: Union[list, None]
+    webhook_configs: Union[list, None]
+    wechat_configs: Union[list, None]
 
 
 @dataclass

tests/integrationtest/test_alerting.py

@@ -6,6 +6,7 @@
 from grafana_api.model import (
     APIModel,
     AlertmanagerConfig,
+    AlertmanagerReceivers,
     DatasourceRuleQuery,
 )
 from grafana_api.alerting import Alerting
@@ -107,40 +108,40 @@ def test_c_delete_alertmanager_config(self):
 
         self.assertEqual(result, self.alerting.get_alertmanager_config())
 
-    # def test_test_alertmanager_receivers(self):
-    #     grafana_managed_receiver_configs: list = [
-    #         {
-    #             "uid": None,
-    #             "name": "email receiver",
-    #             "type": "email",
-    #             "disableResolveMessage": False,
-    #             "settings": {"addresses": "[email protected]"},
-    #             "secureFields": {},
-    #         }
-    #     ]
-    #
-    #     alertmangager_receivers: AlertmanagerReceivers = AlertmanagerReceivers(
-    #         name="Test",
-    #         email_configs=None,
-    #         grafana_managed_receiver_configs=grafana_managed_receiver_configs,
-    #         opsgenie_configs=None,
-    #         pagerduty_configs=None,
-    #         pushover_configs=None,
-    #         slack_configs=None,
-    #         sns_configs=None,
-    #         victorops_configs=None,
-    #         webhook_configs=None,
-    #         wechat_configs=None,
-    #     )
-    #     self.assertEqual(
-    #         None,
-    #         self.alerting.test_alertmanager_receivers(
-    #             alert=dict(
-    #                 {"annotations": {"test": "test"}, "labels": {"test": "test"}}
-    #             ),
-    #             receivers=list([alertmangager_receivers]),
-    #         ),
-    #     )
+    def test_test_alertmanager_receivers(self):
+        grafana_managed_receiver_configs: list = [
+            {
+                "uid": None,
+                "name": "email receiver",
+                "type": "email",
+                "disableResolveMessage": False,
+                "settings": {"addresses": "[email protected]"},
+                "secureFields": {},
+            }
+        ]
+
+        alertmangager_receivers: AlertmanagerReceivers = AlertmanagerReceivers(
+            name="Test",
+            email_configs=None,
+            grafana_managed_receiver_configs=grafana_managed_receiver_configs,
+            opsgenie_configs=None,
+            pagerduty_configs=None,
+            pushover_configs=None,
+            slack_configs=None,
+            sns_configs=None,
+            victorops_configs=None,
+            webhook_configs=None,
+            wechat_configs=None,
+        )
+        self.assertEqual(
+            None,
+            self.alerting.test_alertmanager_receivers(
+                alert=dict(
+                    {"annotations": {"test": "test"}, "labels": {"test": "test"}}
+                ),
+                receivers=list([alertmangager_receivers]),
+            ),
+        )
 
     def test_get_prometheus_alerts(self):
         MAX_TRIES: int = 3
@@ -190,19 +191,4 @@ def test_test_rule(self):
         data_queries: list = list([datasource_rule_query])
 
         with self.assertRaises(Exception):
-            self.alerting.test_rule(data_queries)
-
-    def test_test_recipient_rule(self):
-        datasource_rule_query: DatasourceRuleQuery = DatasourceRuleQuery(
-            datasource_uid="test",
-            model=dict(),
-            query_type="",
-            ref_id="test1",
-            relative_time_range=dict({"from": 20, "to": 90}),
-        )
-        data_queries: list = list([datasource_rule_query])
-
-        with self.assertRaises(Exception):
-            self.alerting.test_recipient_rule(
-                expr="test", condition="test", data_query=data_queries
-            )
+            self.alerting.test_rule(data_queries)