changes based on Johan's feedback

src/cnaas_nms/api/auth.py

@@ -12,7 +12,7 @@
 from cnaas_nms.app_settings import auth_settings
 from cnaas_nms.tools.log import get_logger
 from cnaas_nms.tools.rbac.rbac import get_permissions_user
-from cnaas_nms.tools.security import get_identity, get_oauth_token_info, login_required, login_required_all_permitted
+from cnaas_nms.tools.security import get_identity, get_oauth_token_info, login_required_all_permitted
 from cnaas_nms.version import __api_version__
 
 logger = get_logger()
@@ -141,7 +141,7 @@ def post(self):
 
 
 class IdentityApi(Resource):
-    @login_required
+    @login_required_all_permitted
     def get(self):
         identity = get_identity()
         return identity
@@ -155,7 +155,7 @@ def get(self):
             logger.debug("No permissions defined, so nobody is permitted to do any api calls.")
             return []
         user_info = get_oauth_token_info(current_token)
-        permissions_of_user = get_permissions_user(permissions_rules, user_info)  # check check
+        permissions_of_user = get_permissions_user(permissions_rules, user_info)
 
         # convert to dictionaries so it can be converted to json
         permissions_as_dics = []

src/cnaas_nms/api/device.py

@@ -483,7 +483,7 @@ def post(self, device_id: int):
 
     @classmethod
     def arg_check(cls, device_id: int, json_data: dict) -> dict:
-        parsed_args: dict[str, Any] = {"device_id": device_id}
+        parsed_args: dict[str, int | str | list[Any] | None] = {"device_id": device_id}
         if not isinstance(device_id, int):
             raise ValueError("'device_id' must be an integer")
 

src/cnaas_nms/devicehandler/sync_devices.py

@@ -879,14 +879,12 @@ def sync_devices(
                 )
             raise Exception("Configuration hash check failed for {}".format(" ".join(nrresult.failed_hosts.keys())))
 
-    if not dry_run:
+    if not dry_run and job_id:
         with sqla_session() as session:  # type: ignore
             logger.info("Trying to acquire lock for devices to run syncto job: {}".format(job_id))
             max_attempts = 5
             lock_ok: bool = False
             for i in range(max_attempts):
-                if not job_id:
-                    continue
                 lock_ok = Joblock.acquire_lock(session, name="devices", job_id=job_id)
                 if lock_ok:
                     break

src/cnaas_nms/tools/dhcp_hook.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import sys
-from typing import Union
+from typing import Optional
 
 import netaddr
 import requests
@@ -55,7 +55,7 @@ def canonical_mac(mac):
     return str(na_mac)
 
 
-def main() -> Union[int, None]:
+def main() -> Optional[int]:
     if len(sys.argv) < 3:
         return 1
 

src/cnaas_nms/tools/jinja_filters.py

@@ -4,7 +4,7 @@
 import hashlib
 import ipaddress
 import re
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional
 
 # This global dict can be used to update the Jinja environment filters dict to include all
 # registered template filter function
@@ -135,9 +135,9 @@ def ipv4_to_ipv6(
 
 @template_filter()
 def get_interface(
-    network: Union[ipaddress.IPv6Interface, ipaddress.IPv4Interface, ipaddress.IPv6Network, ipaddress.IPv4Network, str],
+    network: ipaddress.IPv6Interface | ipaddress.IPv4Interface | ipaddress.IPv6Network | ipaddress.IPv4Network | str,
     index: int,
-) -> Union[ipaddress.IPv6Interface, ipaddress.IPv4Interface, ipaddress.IPv6Network, ipaddress.IPv4Network]:
+) -> ipaddress.IPv6Interface | ipaddress.IPv4Interface | ipaddress.IPv6Network | ipaddress.IPv4Network:
     """Returns a host address with a prefix length from its index in a network.
 
     Example: