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

Commit

Permalink
make changes to existing classes
Browse files Browse the repository at this point in the history
  • Loading branch information
@nikhilkp93
nikhilkp93 committed Nov 1, 2024
1 parent da44485 commit 2497f69
Showing 1 changed file with 8 additions and 152 deletions.
160 changes: 8 additions & 152 deletions catalystwan/api/mtt_aaa_api.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,6 @@ class TenantRadiusAPI:
def __init__(self, session: ManagerSession) -> None:
self.session = session
self.url_path = "/dataservice/admin/radius"
self.tenant_id = self.session.get_tenant_id()

def __str__(self) -> str:
return str(self.session)
Expand All @@ -113,7 +112,7 @@ def add_radius(self, radius_server: TenantRadiusServer):
:param radius_server:
:return:
"""
logger.debug(f"Add RADIUS config tenant_id={self.tenant_id}.")
logger.debug("Add RADIUS config")
data = asdict(radius_server) # type: ignore
return self.session.post(url=self.url_path, json=data)

Expand All @@ -124,7 +123,7 @@ def put_radius(self, radius_server: TenantRadiusServer):
:param radius_server:
:return:
"""
logger.debug(f"Update RADIUS config tenant_id={self.tenant_id}.")
logger.debug("Update RADIUS config")
data = asdict(radius_server) # type: ignore
return self.session.put(url=self.url_path, json=data)

Expand All @@ -135,86 +134,15 @@ def delete_radius(self):
:param radius_server:
:return: True|False
"""
logger.debug(f"Delete RADIUS config tenant_id={self.tenant_id}.")
logger.debug("Delete RADIUS config")
return self.session.delete(self.url_path)

def get_radius(self) -> TenantRadiusServer:
"""
Retrieve Radius server
:return: TenantRadiusServer
"""
logger.debug(f"RADIUS config tenant_id={self.tenant_id}.")
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.")
logger.debug("RADIUS config")
data = self.session.get_data(self.url_path)
return create_dataclass(TenantRadiusServer, data)

Expand All @@ -227,7 +155,6 @@ class TenantTacacsAPI:
def __init__(self, session: ManagerSession) -> None:
self.session = session
self.url_path = "/dataservice/admin/tacacs"
self.tenant_id = self.session.get_tenant_id()

def __str__(self) -> str:
return str(self.session)
Expand All @@ -239,7 +166,7 @@ def add_tacacs(self, tacacs_server: TenantTacacsServer):
:param tacacs_server:
:return:
"""
logger.debug(f"TACACS config tenant_id={self.tenant_id}.")
logger.debug("TACACS config")
data = asdict(tacacs_server) # type: ignore
return self.session.post(url=self.url_path, json=data)

Expand All @@ -250,7 +177,7 @@ def put_tacacs(self, tacacs_server: TenantTacacsServer):
:param tacacs_server:
:return:
"""
logger.debug(f"Update TACACS config tenant_id={self.tenant_id}.")
logger.debug("Update TACACS config")
data = asdict(tacacs_server) # type: ignore
return self.session.put(url=self.url_path, json=data)

Expand All @@ -261,85 +188,14 @@ def delete_tacacs(self):
:param tacacs_server:
:return: True|False
"""
logger.debug(f"Delete TACACS config tenant_id={self.tenant_id}.")
logger.debug("Delete TACACS config")
return self.session.delete(self.url_path)

def get_tacacs(self) -> TenantTacacsServer:
"""
Retrieves Tacacs server
:return: 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.")
logger.debug("TACACS config")
data = self.session.get_data(self.url_path)
return create_dataclass(TenantTacacsServer, data)

0 comments on commit 2497f69

Please sign in to comment.