Development dependencies (#12)

.flake8

@@ -0,0 +1,13 @@
+[flake8]
+exclude = .venv,.git,.tox,docs,venv,bin,lib,deps,build
+# To work with Black
+max-line-length = 88
+# D202 No blank lines allowed after function docstring
+# E203: Whitespace before ':'
+# E501: line too long
+# W503: Line break occurred before a binary operator
+ignore =
+    D202,
+    E203,
+    E501,
+    W503

.gitignore

@@ -1,7 +1,9 @@
 nac_collector/__pycache__
-*.yaml
 *.json
 tmp/
 .envrc
 .DS_Store
 .pylintrc
+
+# pyenv
+.python-version

.mypy.ini

@@ -0,0 +1,16 @@
+[mypy]
+# The mypy configurations: http://bit.ly/2zEl9WI
+python_version = 3.6
+check_untyped_defs = True
+disallow_any_generics = True
+disallow_untyped_calls = True
+disallow_untyped_defs = True
+disallow_incomplete_defs = True
+disallow_untyped_decorators = False
+ignore_errors = False
+ignore_missing_imports = True
+strict_optional = True
+warn_unused_configs = True
+warn_unused_ignores = False
+warn_return_any = True
+warn_redundant_casts = True

.pre-commit-config.yaml

@@ -0,0 +1,16 @@
+repos:
+  - repo: https://github.com/psf/black
+    rev: 24.4.2
+    hooks:
+      - id: black
+        args:
+          - --safe
+          - --quiet
+  - repo: https://github.com/pycqa/flake8
+    rev: 5.0.4
+    hooks:
+      - id: flake8
+  - repo: https://github.com/timothycrosley/isort
+    rev: 5.13.2
+    hooks:
+      - id: isort

nac_collector/cisco_client.py

@@ -1,10 +1,11 @@
-import time
-import logging
-import json
 from abc import ABC, abstractmethod
+import json
+import logging
+import time
+
 import requests
-import urllib3
 from ruamel.yaml import YAML
+import urllib3
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
@@ -89,9 +90,13 @@ def get_request(self, url):
         for _ in range(self.max_retries):
             try:
                 # Send a GET request to the URL
-                response = self.session.get(url, verify=self.ssl_verify, timeout=self.timeout)
+                response = self.session.get(
+                    url, verify=self.ssl_verify, timeout=self.timeout
+                )
             except requests.exceptions.Timeout:
-                self.logger.error("GET %s timed out after %s seconds.", url, self.timeout)
+                self.logger.error(
+                    "GET %s timed out after %s seconds.", url, self.timeout
+                )
                 continue
 
             if response.status_code == 429:
@@ -134,9 +139,13 @@ def post_request(self, url, data):
         for _ in range(self.max_retries):
             try:
                 # Send a POST request to the URL
-                response = self.session.post(url, data=data, verify=self.ssl_verify, timeout=self.timeout)
+                response = self.session.post(
+                    url, data=data, verify=self.ssl_verify, timeout=self.timeout
+                )
             except requests.exceptions.Timeout:
-                self.logger.error("POST %s timed out after %s seconds.", url, self.timeout)
+                self.logger.error(
+                    "POST %s timed out after %s seconds.", url, self.timeout
+                )
                 continue
 
             if response.status_code == 429:

nac_collector/cisco_client_ise.py

@@ -1,4 +1,5 @@
 import logging
+
 import requests
 import urllib3
 
@@ -33,7 +34,9 @@ def __init__(
         timeout,
         ssl_verify,
     ):
-        super().__init__(username, password, base_url, max_retries, retry_after, timeout, ssl_verify)
+        super().__init__(
+            username, password, base_url, max_retries, retry_after, timeout, ssl_verify
+        )
 
     def authenticate(self):
         """
@@ -73,7 +76,9 @@ def authenticate(self):
                 self.session = requests.Session()
                 self.session.auth = (self.username, self.password)
                 self.session.headers.update(headers)
-                self.session.headers.update({"Content-Type": "application/json", "Accept": "application/json"})
+                self.session.headers.update(
+                    {"Content-Type": "application/json", "Accept": "application/json"}
+                )
                 return True
 
             logger.error(
@@ -115,14 +120,18 @@ def get_from_endpoints(self, endpoints_yaml_file):
                 data = response.json()
                 # License API returns a list of dictionaries
                 if isinstance(data, list):
-                    endpoint_dict[endpoint["name"]].append({"data": data, "endpoint": endpoint["endpoint"]})
+                    endpoint_dict[endpoint["name"]].append(
+                        {"data": data, "endpoint": endpoint["endpoint"]}
+                    )
 
                 elif data.get("response"):
                     for i in data.get("response"):
                         endpoint_dict[endpoint["name"]].append(
                             {
                                 "data": i,
-                                "endpoint": endpoint["endpoint"] + "/" + self.get_id_value(i),
+                                "endpoint": endpoint["endpoint"]
+                                + "/"
+                                + self.get_id_value(i),
                             }
                         )
                 # Pagination for ERS API results
@@ -132,7 +141,9 @@ def get_from_endpoints(self, endpoints_yaml_file):
                         endpoint_dict[endpoint["name"]].append(
                             {
                                 "data": i,
-                                "endpoint": endpoint["endpoint"] + "/" + self.get_id_value(i),
+                                "endpoint": endpoint["endpoint"]
+                                + "/"
+                                + self.get_id_value(i),
                             }
                         )
                 # Check if response is empty list
@@ -156,7 +167,9 @@ def get_from_endpoints(self, endpoints_yaml_file):
                 index = 0
                 # Iterate over the items in final_dict[parent_endpoint]
                 for item in value:
-                    if parent_endpoint == "/".join(item.get("endpoint").split("/")[:-1]):
+                    if parent_endpoint == "/".join(
+                        item.get("endpoint").split("/")[:-1]
+                    ):
                         # Initialize an empty list for parent endpoint ids
                         parent_endpoint_ids = []
 
@@ -188,7 +201,9 @@ def get_from_endpoints(self, endpoints_yaml_file):
                                     value[index]["children"][endpoint["name"]].append(
                                         {
                                             "data": i,
-                                            "endpoint": new_endpoint + "/" + self.get_id_value(i),
+                                            "endpoint": new_endpoint
+                                            + "/"
+                                            + self.get_id_value(i),
                                         }
                                     )
                             self.log_response(new_endpoint, response)