fix: pylint

nautobot_chatops/dispatchers/base.py

@@ -53,7 +53,7 @@ def user(self):
                 ).nautobot_user
             except ObjectDoesNotExist:
                 logger.warning(
-                    "Could not find User matching %s - id: %s." "Add a ChatOps User to link the accounts.",
+                    "Could not find User matching %s - id: %s. Add a ChatOps User to link the accounts.",
                     self.context["user_name"],
                     self.context["user_id"],
                 )
@@ -240,7 +240,6 @@ def multi_input_dialog(self, command, sub_command, dialog_title, dialog_list):
         """
         raise NotImplementedError
 
-    # pylint: disable=no-self-use
     def needs_permission_to_send_image(self):
         """Return True if this bot needs to ask the user for permission to post an image."""
         return False
@@ -326,17 +325,14 @@ def user_mention(self):
         """Markup for a mention of the username/userid specified in our context."""
         raise NotImplementedError
 
-    # pylint: disable=no-self-use
     def bold(self, text):
         """Mark text as bold."""
         return f"**{text}**"
 
-    # pylint: disable=no-self-use
     def hyperlink(self, text, url):
         """Create Hyperlinks."""
         return f"[{text}]({url})"
 
-    # pylint: disable=no-self-use
     def monospace(self, text):
         """Mark text as monospace."""
         return f"`{text}`"

nautobot_chatops/dispatchers/mattermost.py

@@ -606,7 +606,6 @@ def actions_block(self, block_id, actions):
         # Leaving in place to pass the testing.
         return {"block_id": block_id, "actions": actions}
 
-    # pylint: disable=no-self-use
     def _input_block(self, block_id, label, element):
         """Construct a block consisting of Input elements."""
         element["display_name"] = label

nautobot_chatops/dispatchers/ms_teams.py

@@ -44,7 +44,7 @@ def user(self):
                 ).nautobot_user
             except ObjectDoesNotExist:
                 logger.warning(
-                    "Could not find User matching %s - id: %s." "Add a ChatOps User to link the accounts.",
+                    "Could not find User matching %s - id: %s. Add a ChatOps User to link the accounts.",
                     self.context["user_name"],
                     self.context["user_ad_id"],
                 )

nautobot_chatops/dispatchers/slack.py

@@ -315,7 +315,6 @@ def send_exception(self, exception):
             text=f"Sorry @{self.context.get('user_name')}, an error occurred :sob:\n```{exception}```",
         )
 
-    # pylint: disable=no-self-use
     def delete_message(self, response_url):
         """Delete a message that was previously sent."""
         WebhookClient(response_url).send_dict({"delete_original": "true"})

nautobot_chatops/integrations/aci/aci.py

@@ -204,6 +204,7 @@ def get_static_path(self, tenant: str, ap: str, epg: str) -> list:
         )
         sp_list = []
         for obj in resp.json()["imdata"]:
+            # pylint: disable-next=use-dict-literal
             sp_dict = dict(encap=obj["fvRsPathAtt"]["attributes"]["encap"])
             # pylint: disable-next=invalid-name
             tDn = obj["fvRsPathAtt"]["attributes"]["tDn"]
@@ -273,6 +274,7 @@ def get_static_path(self, tenant: str, ap: str, epg: str) -> list:
     def get_epg_details(self, tenant: str, ap: str, epg: str) -> dict:
         """Return EPG configuration details."""
         resp = self._get(f"/api/node/mo/uni/tn-{tenant}/ap-{ap}/epg-{epg}.json?query-target=children")
+        # pylint: disable-next=use-dict-literal
         epg_dict = dict(bd=None, subnets=[], provided_contracts=[], consumed_contracts=[], domains=[], static_paths=[])
         epg_dict["name"] = epg
         for obj in resp.json()["imdata"]:
@@ -281,13 +283,15 @@ def get_epg_details(self, tenant: str, ap: str, epg: str) -> dict:
                 epg_dict["subnets"] = self.get_bd_subnet(tenant, epg_dict["bd"])
             if "fvRsCons" in obj:
                 epg_dict["consumed_contracts"].append(
+                    # pylint: disable-next=use-dict-literal
                     dict(
                         name=obj["fvRsCons"]["attributes"]["tnVzBrCPName"],
                         filters=self.get_contract_filters(tenant, obj["fvRsCons"]["attributes"]["tnVzBrCPName"]),
                     )
                 )
             if "fvRsProv" in obj:
                 epg_dict["provided_contracts"].append(
+                    # pylint: disable-next=use-dict-literal
                     dict(
                         name=obj["fvRsProv"]["attributes"]["tnVzBrCPName"],
                         filters=self.get_contract_filters(tenant, obj["fvRsProv"]["attributes"]["tnVzBrCPName"]),

nautobot_chatops/integrations/ansible/tower.py

@@ -10,6 +10,8 @@
 
 _CONFIG = settings.PLUGINS_CONFIG["nautobot_chatops"]
 
+DEFAULT_TIMEOUT = 20
+
 
 def _get_uri(uri):
     """Validate URI schema and no trailing slash.
@@ -79,6 +81,7 @@ def _launch_job(self, template_name, extra_vars):
             headers=self.headers,
             data=json.dumps({"extra_vars": extra_vars}),
             verify=self.tower_verify_ssl,  # nosec
+            timeout=DEFAULT_TIMEOUT,
         )
         response.raise_for_status()
         logger.info("Job submission to Ansible Tower:")
@@ -100,6 +103,7 @@ def _get_tower(self, api_path, **kwargs):
             auth=(self.username, self.password),
             **kwargs,
             verify=self.tower_verify_ssl,  # nosec
+            timeout=DEFAULT_TIMEOUT,
         )
         return response.json()
 

nautobot_chatops/integrations/aristacv/cvpgrpcutils.py

@@ -25,7 +25,10 @@ def connect_cv(settings):
             cert = bytes(ssl.get_server_certificate((cvp_host, 8443)), "utf-8")
             channel_creds = grpc.ssl_channel_credentials(cert)
             response = requests.post(
-                f"https://{cvp_host}/cvpservice/login/authenticate.do", auth=(username, password), verify=False  # nosec
+                f"https://{cvp_host}/cvpservice/login/authenticate.do",
+                auth=(username, password),
+                verify=False,  # nosec
+                timeout=DEFAULT_TIMEOUT,
             )
         # Otherwise, the server is expected to have a valid certificate signed by a well-known CA.
         else:

nautobot_chatops/integrations/aristacv/utils.py

@@ -560,6 +560,7 @@ def get_token_crt():
             f"https://{CVP_HOST}/cvpservice/login/authenticate.do",
             auth=(CVP_USERNAME, CVP_PASSWORD),
             verify=False,  # nosec
+            timeout=DEFAULT_TIMEOUT,
         )
     else:
         request = requests.post(

nautobot_chatops/integrations/grafana/filters.py

@@ -18,7 +18,7 @@ class Meta:
 
         fields = ("dashboard_slug", "dashboard_uid", "friendly_name")
 
-    def search(self, queryset, name, value):  # pylint: disable=unused-argument,no-self-use
+    def search(self, queryset, name, value):  # pylint: disable=unused-argument
         """Perform the filtered search."""
         if not value.strip():
             return queryset
@@ -40,7 +40,7 @@ class Meta:
 
         fields = ("dashboard", "command_name", "friendly_name", "panel_id")
 
-    def search(self, queryset, name, value):  # pylint: disable=unused-argument,no-self-use
+    def search(self, queryset, name, value):  # pylint: disable=unused-argument
         """Perform the filtered search."""
         if not value.strip():
             return queryset
@@ -66,7 +66,7 @@ class Meta:
 
         fields = ("panel", "name", "friendly_name", "query", "modelattr", "value", "response")
 
-    def search(self, queryset, name, value):  # pylint: disable=unused-argument,no-self-use
+    def search(self, queryset, name, value):  # pylint: disable=unused-argument
         """Perform the filtered search."""
         if not value.strip():
             return queryset

nautobot_chatops/integrations/meraki/worker.py

@@ -28,6 +28,7 @@
 except KeyError as err:
     MERAKI_DASHBOARD_API_KEY = os.getenv("MERAKI_DASHBOARD_API_KEY")
     if not MERAKI_DASHBOARD_API_KEY:
+        # pylint: disable-next=broad-exception-raised
         raise Exception("Unable to find the Meraki API key.") from err
 
 
@@ -570,6 +571,7 @@ def configure_basic_access_port(  # pylint: disable=too-many-arguments
             dialog_list=dialog_list,
         )
         return False
+    # pylint: disable-next=use-dict-literal
     port_params = dict(name=port_desc, enabled=bool(enabled), type="access", vlan=vlan)
     LOGGER.info("PORT PARMS: %s", port_params)
     client = MerakiClient(api_key=MERAKI_DASHBOARD_API_KEY)

nautobot_chatops/integrations/panorama/constant.py

@@ -22,3 +22,4 @@
 PANOS_MANUFACTURER_NAME = "Palo Alto Networks"
 PANOS_PLATFORM = "PANOS"
 PANOS_DEVICE_ROLE = "Firewall"
+DEFAULT_TIMEOUT = 20

nautobot_chatops/integrations/panorama/utils.py

@@ -12,7 +12,7 @@
 from panos.policies import PostRulebase, PreRulebase, Rulebase, SecurityRule
 from requests.exceptions import RequestException
 
-from .constant import PLUGIN_CFG
+from .constant import DEFAULT_TIMEOUT, PLUGIN_CFG
 
 
 logger = logging.getLogger(__name__)
@@ -31,7 +31,12 @@ def get_api_key_api(url: str = PLUGIN_CFG["panorama_host"]) -> str:
 
     params = {"type": "keygen", "user": PLUGIN_CFG["panorama_user"], "password": PLUGIN_CFG["panorama_password"]}
 
-    response = requests.get(f"https://{url}/api/", params=params, verify=False)  # nosec
+    response = requests.get(
+        f"https://{url}/api/",
+        params=params,
+        verify=False,  # nosec
+        timeout=DEFAULT_TIMEOUT,
+    )
     if response.status_code != 200:
         raise RequestException(f"Something went wrong while making a request. Reason: {response.text}")
 
@@ -218,7 +223,12 @@ def _get_pcap(capture_filename: str, ip_address: str):
 
     params = {"key": get_api_key_api(), "type": "export", "category": "filters-pcap", "from": "1.pcap"}
 
-    respone = requests.get(url, params=params, verify=False)  # nosec
+    respone = requests.get(
+        url,
+        params=params,
+        verify=False,  # nosec
+        timeout=DEFAULT_TIMEOUT,
+    )
 
     with open(capture_filename, "wb") as pcap_file:
         pcap_file.write(respone.content)

nautobot_chatops/sockets/slack.py

@@ -15,11 +15,11 @@
 from nautobot_chatops.utils import socket_check_and_enqueue_command
 
 
-async def main():  # pylint: disable=too-many-statements
+# pylint: disable-next=too-many-statements
+async def main():
     """Slack Socket Main Loop."""
-    SLASH_PREFIX = settings.PLUGINS_CONFIG["nautobot_chatops"].get(  # pylint:disable=invalid-name
-        "slack_slash_command_prefix"
-    )
+    # pylint: disable-next=invalid-name
+    SLASH_PREFIX = settings.PLUGINS_CONFIG["nautobot_chatops"].get("slack_slash_command_prefix")
     client = SocketModeClient(
         app_token=settings.PLUGINS_CONFIG["nautobot_chatops"].get("slack_app_token"),
         web_client=AsyncWebClient(token=settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_api_token"]),
@@ -78,7 +78,7 @@ async def process_slash_command(client, req):
 
         return await socket_check_and_enqueue_command(registry, command, subcommand, params, context, SlackDispatcher)
 
-    # pylint: disable=too-many-locals,too-many-return-statements,too-many-branches,too-many-statements,too-many-nested-blocks
+    # pylint: disable-next=too-many-locals,too-many-return-statements,too-many-branches,too-many-statements
     async def process_interactive(client, req):
         client.logger.debug("Processing interactive.")
         payload = req.payload
@@ -125,6 +125,7 @@ async def process_interactive(client, req):
                 # Nothing more to do
                 return
 
+        # pylint: disable-next=too-many-nested-blocks
         elif "view" in payload and payload["view"]:
             # View submission triggered from a modal dialog
             client.logger.info("Submission triggered from a modal dialog")

nautobot_chatops/tests/aci/test_aci.py

@@ -1,5 +1,5 @@
 """Tests for integrations.aci.aci."""
-# pylint: disable=no-self-use, import-outside-toplevel, invalid-name
+# pylint: disable=invalid-name
 import unittest
 from unittest.mock import patch, Mock
 from nautobot_chatops.integrations.aci.aci import NautobotPluginChatopsAci, RequestHTTPError

nautobot_chatops/tests/meraki/test_utils.py

@@ -9,7 +9,7 @@ class TestUtils(unittest.TestCase):
     """Test Version is the same."""
 
     @patch("nautobot_chatops.integrations.meraki.utils.MerakiClient.get_meraki_orgs")
-    def test_org_name_to_id(self, mock_orgs):  # pylint: disable=no-self-use
+    def test_org_name_to_id(self, mock_orgs):
         """Test Translate Org Name to Org Id."""
         mock_orgs.return_value = [
             {
@@ -23,7 +23,7 @@ def test_org_name_to_id(self, mock_orgs):  # pylint: disable=no-self-use
         assert client.org_name_to_id("NTC-TEST") == "123456"
 
     @patch("nautobot_chatops.integrations.meraki.utils.MerakiClient.get_meraki_devices")
-    def test_name_to_serial(self, mock_devices):  # pylint: disable=no-self-use
+    def test_name_to_serial(self, mock_devices):
         """Test Translate Name to Serial."""
         mock_devices.return_value = [
             {
@@ -45,7 +45,7 @@ def test_name_to_serial(self, mock_devices):  # pylint: disable=no-self-use
         assert client.name_to_serial("NTC-TEST", "fw01-test") == "SN123456"
 
     @patch("nautobot_chatops.integrations.meraki.utils.MerakiClient.get_meraki_networks_by_org")
-    def test_netname_to_id(self, mock_net_name):  # pylint: disable=no-self-use
+    def test_netname_to_id(self, mock_net_name):
         """Translate Network Name to Network ID."""
         mock_net_name.return_value = [
             {

nautobot_chatops/tests/test_utils.py

@@ -144,7 +144,6 @@ def test_default_deny(self, mock_enqueue_task):
         self.assertIsNone(MockDispatcher.error)
         mock_enqueue_task.assert_called_once()
 
-    # pylint: disable=no-self-use
     def setup_db(self):
         """Per-testcase database population for most test cases."""
         # Create some globally applicable access grants: