Skip to content
This repository has been archived by the owner on Nov 21, 2024. It is now read-only.

Commit

Permalink
add support for tenant level API calls for mtt aaa
Browse files Browse the repository at this point in the history
  • Loading branch information
nikhilkp93 committed Oct 31, 2024
1 parent b4916c5 commit 2aef5d9
Showing 1 changed file with 141 additions and 0 deletions.
141 changes: 141 additions & 0 deletions catalystwan/api/mtt_aaa_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,76 @@ def get_radius(self) -> TenantRadiusServer:
data = self.session.get_data(self.url_path)
return create_dataclass(TenantRadiusServer, data)

class TenantLevelRadiusAPI:
"""
Class to configure tenant remote AAA RADIUS servers as tenant level users.
"""

def __init__(self, session: ManagerSession) -> None:
self.session = session
self.url_path = "/dataservice/admin/radius"

def __str__(self) -> str:
return str(self.session)

@status_ok
def add_radius(self, radius_server: TenantRadiusServer):
"""
Create RADIUS server for tenant.
:param radius_server: TenantRadiusServer object containing the server configuration.
:return: True if successful, False otherwise.
"""
logger.debug("Adding RADIUS server.")
data = asdict(radius_server) # type: ignore
try:
response = self.session.post(url=self.url_path, json=data)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Error adding RADIUS server: {e}")
return False

@status_ok
def put_radius(self, radius_server: TenantRadiusServer):
"""
Edit RADIUS server for tenant.
:param radius_server: TenantRadiusServer object containing the server configuration.
:return: True if successful, False otherwise.
"""
logger.debug("Updating RADIUS server.")
data = asdict(radius_server) # type: ignore
try:
response = self.session.put(url=self.url_path, json=data)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Error updating RADIUS server: {e}")
return False

@status_ok
def delete_radius(self):
"""
Delete RADIUS server for tenant.
:return: True if successful, False otherwise.
"""
logger.debug("Deleting RADIUS server.")
try:
response = self.session.delete(self.url_path)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Error deleting RADIUS server: {e}")
return False

def get_radius(self) -> TenantRadiusServer:
"""
Retrieve RADIUS server configuration for tenant.
:return: TenantRadiusServer object containing the server configuration.
"""
logger.debug("Retrieving RADIUS server configuration.")
data = self.session.get_data(self.url_path)
return create_dataclass(TenantRadiusServer, data)


class TenantTacacsAPI:
"""
Expand Down Expand Up @@ -201,3 +271,74 @@ def get_tacacs(self) -> TenantTacacsServer:
logger.debug(f"TACACS config tenant_id={self.tenant_id}.")
data = self.session.get_data(self.url_path)
return create_dataclass(TenantTacacsServer, data)


class TenantLevelTacacsAPI:
"""
Class to configure tenant remote AAA TACACS servers as tenant level user.
"""

def __init__(self, session: ManagerSession) -> None:
self.session = session
self.url_path = "/dataservice/admin/tacacs"

def __str__(self) -> str:
return str(self.session)

@status_ok
def add_tacacs(self, tacacs_server: TenantTacacsServer):
"""
Create TACACS server for tenant.
:param tacacs_server: TenantTacacsServer object containing the server configuration.
:return: True if successful, False otherwise.
"""
logger.debug("Adding TACACS server.")
data = asdict(tacacs_server) # type: ignore
try:
response = self.session.post(url=self.url_path, json=data)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Error adding TACACS server: {e}")
return False

@status_ok
def put_tacacs(self, tacacs_server: TenantTacacsServer):
"""
Update TACACS server for tenant.
:param tacacs_server: TenantTacacsServer object containing the server configuration.
:return: True if successful, False otherwise.
"""
logger.debug("Updating TACACS server.")
data = asdict(tacacs_server) # type: ignore
try:
response = self.session.put(url=self.url_path, json=data)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Error updating TACACS server: {e}")
return False

@status_ok
def delete_tacacs(self):
"""
Delete TACACS server for tenant.
:return: True if successful, False otherwise.
"""
logger.debug("Deleting TACACS server.")
try:
response = self.session.delete(self.url_path)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Error deleting TACACS server: {e}")
return False

def get_tacacs(self) -> TenantTacacsServer:
"""
Retrieve TACACS server configuration for tenant.
:return: TenantTacacsServer object containing the server configuration.
"""
logger.debug("Retrieving TACACS server configuration.")
data = self.session.get_data(self.url_path)
return create_dataclass(TenantTacacsServer, data)

0 comments on commit 2aef5d9

Please sign in to comment.