From a1ef5f687275707bda06a4e81a23cfcf509d0aa5 Mon Sep 17 00:00:00 2001 From: Dario Kaelin Date: Wed, 3 Dec 2025 22:14:05 +0100 Subject: [PATCH] Release R1.7 --- .github/workflows/python-app.yml | 2 +- CHANGELOG.md | 3 +- aciClient/__init__.py | 5 +- aciClient/aci.py | 162 ++++++----- aciClient/aciCertClient.py | 48 ++-- requirements.txt | 11 +- setup.py | 34 ++- test/test_aci.py | 461 +++++++++++++++++++------------ 8 files changed, 439 insertions(+), 287 deletions(-) diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 612144f..05f8369 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -15,7 +15,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ["3.8","3.9", "3.10", "3.11", "3.12"] + python-version: ["3.10", "3.11", "3.12", "3.13"] steps: - uses: actions/checkout@v4 diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ae2739..40bd0d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,4 +3,5 @@ > * v1.00, 11.2020 -- Adjustments for OpenSource - Andreas Graber > * v1.3, 06.2021 -- Snapshot creation added - Dario Kaelin > * v1.5, 01.2023 -- Dependency Update -> * v1.6, 06.2024 -- Improvement for subscription - Dario Kaelin \ No newline at end of file +> * v1.6, 06.2024 -- Improvement for subscription - Dario Kaelin +> * v1.7, 12.2025 -- Dependency Update - Dario Kaelin \ No newline at end of file diff --git a/aciClient/__init__.py b/aciClient/__init__.py index 3162d84..41a10a4 100644 --- a/aciClient/__init__.py +++ b/aciClient/__init__.py @@ -1,7 +1,4 @@ from aciClient.aci import ACI from aciClient.aciCertClient import ACICert -__all__ = [ - 'ACI', - 'ACICert' -] +__all__ = ["ACI", "ACICert"] diff --git a/aciClient/aci.py b/aciClient/aci.py index 0a74536..12bfdde 100644 --- a/aciClient/aci.py +++ b/aciClient/aci.py @@ -32,14 +32,14 @@ class ACI: # constructor # ============================================================================== def __init__(self, apicIp, apicUser, apicPasword, refresh=False, proxies=None): - self.__logger.debug('Constructor called') + self.__logger.debug("Constructor called") self.apicIp = apicIp self.apicUser = apicUser self.apicPassword = apicPasword self.proxies = proxies - self.baseUrl = 'https://' + self.apicIp + '/api/' - self.__logger.debug(f'BaseUrl set to: {self.baseUrl}') + self.baseUrl = "https://" + self.apicIp + "/api/" + self.__logger.debug(f"BaseUrl set to: {self.baseUrl}") self.refresh_auto = refresh self.refresh_next = None @@ -52,17 +52,27 @@ def __init__(self, apicIp, apicUser, apicPasword, refresh=False, proxies=None): self.retry_backoff_factor = 10 # in seconds; multiplied by previous attempts. def __refresh_session_timer(self, response): - self.__logger.debug(f'refreshing the token {self.refresh_offset}s before it expires') - self.refresh_next = int(response.json()['imdata'][0]['aaaLogin']['attributes']['refreshTimeoutSeconds']) - self.refresh_thread = threading.Timer(self.refresh_next - self.refresh_offset, self.renewCookie) - self.__logger.debug(f'starting thread to refresh token in {self.refresh_next - self.refresh_offset}s') + self.__logger.debug( + f"refreshing the token {self.refresh_offset}s before it expires" + ) + self.refresh_next = int( + response.json()["imdata"][0]["aaaLogin"]["attributes"][ + "refreshTimeoutSeconds" + ] + ) + self.refresh_thread = threading.Timer( + self.refresh_next - self.refresh_offset, self.renewCookie + ) + self.__logger.debug( + f"starting thread to refresh token in {self.refresh_next - self.refresh_offset}s" + ) self.refresh_thread.start() # ============================================================================== # login # ============================================================================== def login(self) -> bool: - self.__logger.debug('login called') + self.__logger.debug("login called") retry_strategy = urllib3.Retry( total=self.total_retry_attempts, @@ -72,30 +82,38 @@ def login(self) -> bool: adapter = HTTPAdapter(max_retries=retry_strategy) self.session = requests.Session() - self.session.mount('http://', adapter) - self.session.mount('https://', adapter) - self.__logger.debug('Session Object Created') + self.session.mount("http://", adapter) + self.session.mount("https://", adapter) + self.__logger.debug("Session Object Created") if self.proxies is not None: self.session.proxies = self.proxies # create credentials structure - userPass = json.dumps({'aaaUser': {'attributes': {'name': self.apicUser, 'pwd': self.apicPassword}}}) + userPass = json.dumps( + { + "aaaUser": { + "attributes": {"name": self.apicUser, "pwd": self.apicPassword} + } + } + ) - self.__logger.info(f'Login to apic {self.baseUrl}') - response = self.session.post(self.baseUrl + 'aaaLogin.json', data=userPass, verify=False, timeout=5) + self.__logger.info(f"Login to apic {self.baseUrl}") + response = self.session.post( + self.baseUrl + "aaaLogin.json", data=userPass, verify=False, timeout=5 + ) # Don't raise an exception for 401 if response.status_code == 401: - self.__logger.error(f'Login not possible due to Error: {response.text}') + self.__logger.error(f"Login not possible due to Error: {response.text}") self.session = False return False # Raise a exception for all other 4xx and 5xx status_codes response.raise_for_status() - self.token = response.json()['imdata'][0]['aaaLogin']['attributes']['token'] - self.__logger.debug('Successful get Token from APIC') + self.token = response.json()["imdata"][0]["aaaLogin"]["attributes"]["token"] + self.__logger.debug("Successful get Token from APIC") if self.refresh_auto: self.__refresh_session_timer(response=response) @@ -105,31 +123,34 @@ def login(self) -> bool: # logout # ============================================================================== def logout(self): - self.__logger.debug('logout called') + self.__logger.debug("logout called") self.refresh_auto = False if self.refresh_thread is not None: if self.refresh_thread.is_alive(): - self.__logger.debug('Stoping refresh_auto thread') + self.__logger.debug("Stoping refresh_auto thread") self.refresh_thread.cancel() - self.postJson(jsonData={'aaaUser': {'attributes': {'name': self.apicUser}}}, url='aaaLogout.json') - self.__logger.debug('Logout from APIC sucessfull') + self.postJson( + jsonData={"aaaUser": {"attributes": {"name": self.apicUser}}}, + url="aaaLogout.json", + ) + self.__logger.debug("Logout from APIC sucessfull") # ============================================================================== # renew cookie (aaaRefresh) # ============================================================================== def renewCookie(self) -> bool: - self.__logger.debug('Renew Cookie called') - response = self.session.post(self.baseUrl + 'aaaRefresh.json', verify=False) + self.__logger.debug("Renew Cookie called") + response = self.session.post(self.baseUrl + "aaaRefresh.json", verify=False) if response.status_code == 200: if self.refresh_auto: self.__refresh_session_timer(response=response) - self.token = response.json()['imdata'][0]['aaaLogin']['attributes']['token'] - self.__logger.debug('Successfuly renewed the token') + self.token = response.json()["imdata"][0]["aaaLogin"]["attributes"]["token"] + self.__logger.debug("Successfuly renewed the token") else: self.token = False self.refresh_auto = False - self.__logger.error(f'Could not renew token. {response.text}') + self.__logger.error(f"Could not renew token. {response.text}") response.raise_for_status() return False return True @@ -138,7 +159,7 @@ def renewCookie(self) -> bool: # getToken # ============================================================================== def getToken(self) -> str: - self.__logger.debug('Get Token called') + self.__logger.debug("Get Token called") return self.token # ============================================================================== @@ -146,31 +167,31 @@ def getToken(self) -> str: # ============================================================================== def getJson(self, uri, subscription=False) -> {}: url = self.baseUrl + uri - self.__logger.debug(f'Get Json called url: {url}') + self.__logger.debug(f"Get Json called url: {url}") if subscription: - url = '{}?subscription=yes'.format(url) + url = "{}?subscription=yes".format(url) response = self.session.get(url, verify=False) if response.ok: responseJson = response.json() - self.__logger.debug(f'Successful get Data from APIC: {responseJson}') + self.__logger.debug(f"Successful get Data from APIC: {responseJson}") if subscription: - subscription_id = responseJson['subscriptionId'] - self.__logger.debug(f'Returning Subscription Id: {subscription_id}') + subscription_id = responseJson["subscriptionId"] + self.__logger.debug(f"Returning Subscription Id: {subscription_id}") return subscription_id - return responseJson['imdata'] + return responseJson["imdata"] elif response.status_code == 400: - resp_text = response.json()['imdata'][0]['error']['attributes']['text'] - self.__logger.error(f'Error 400 during get occured: {resp_text}') - if resp_text == 'Unable to process the query, result dataset is too big': + resp_text = response.json()["imdata"][0]["error"]["attributes"]["text"] + self.__logger.error(f"Error 400 during get occured: {resp_text}") + if resp_text == "Unable to process the query, result dataset is too big": # Dataset was too big, we try to grab all the data with pagination - self.__logger.debug(f'Trying with Pagination, uri: {uri}') + self.__logger.debug(f"Trying with Pagination, uri: {uri}") return self.getJsonPaged(uri) return resp_text else: - self.__logger.error(f'Error during get occured: {response.json()}') + self.__logger.error(f"Error during get occured: {response.json()}") return response.json() # ============================================================================== @@ -178,7 +199,7 @@ def getJson(self, uri, subscription=False) -> {}: # ============================================================================== def getJsonPaged(self, uri) -> {}: url = self.baseUrl + uri - self.__logger.debug(f'Get Json Pagination called url: {url}') + self.__logger.debug(f"Get Json Pagination called url: {url}") parsed_url = urlparse(url) parsed_query = parse_qsl(parsed_url.query) @@ -186,44 +207,59 @@ def getJsonPaged(self, uri) -> {}: page = 0 while True: - parsed_query.extend([('page', page), ('page-size', '50000')]) + parsed_query.extend([("page", page), ("page-size", "50000")]) page += 1 - url_to_call = urlunparse((parsed_url[0], parsed_url[1], parsed_url[2], parsed_url[3], - urlencode(parsed_query), parsed_url[5])) + url_to_call = urlunparse( + ( + parsed_url[0], + parsed_url[1], + parsed_url[2], + parsed_url[3], + urlencode(parsed_query), + parsed_url[5], + ) + ) response = self.session.get(url_to_call, verify=False) if response.ok: responseJson = response.json() - self.__logger.debug(f'Successful get Data from APIC: {responseJson}') - if responseJson['imdata']: - return_data.extend(responseJson['imdata']) + self.__logger.debug(f"Successful get Data from APIC: {responseJson}") + if responseJson["imdata"]: + return_data.extend(responseJson["imdata"]) else: return return_data elif response.status_code == 400: - resp_text = '400: ' + response.json()['imdata'][0]['error']['attributes']['text'] - self.__logger.error(f'Error 400 during get occured: {resp_text}') + resp_text = ( + "400: " + + response.json()["imdata"][0]["error"]["attributes"]["text"] + ) + self.__logger.error(f"Error 400 during get occured: {resp_text}") return resp_text else: - self.__logger.error(f'Error during get occured: {response.json()}') + self.__logger.error(f"Error during get occured: {response.json()}") return False # ============================================================================== # postJson # ============================================================================== - def postJson(self, jsonData, url='mo.json') -> {}: - self.__logger.debug(f'Post Json called data: {jsonData}') - response = self.session.post(self.baseUrl + url, verify=False, data=json.dumps(jsonData, sort_keys=True)) + def postJson(self, jsonData, url="mo.json") -> {}: + self.__logger.debug(f"Post Json called data: {jsonData}") + response = self.session.post( + self.baseUrl + url, verify=False, data=json.dumps(jsonData, sort_keys=True) + ) if response.status_code == 200: - self.__logger.debug(f'Successful Posted Data to APIC: {response.json()}') + self.__logger.debug(f"Successful Posted Data to APIC: {response.json()}") return response.status_code elif response.status_code == 400: - resp_text = '400: ' + response.json()['imdata'][0]['error']['attributes']['text'] - self.__logger.error(f'Error 400 during get occured: {resp_text}') + resp_text = ( + "400: " + response.json()["imdata"][0]["error"]["attributes"]["text"] + ) + self.__logger.error(f"Error 400 during get occured: {resp_text}") return resp_text else: - self.__logger.error(f'Error during get occured: {response.json()}') + self.__logger.error(f"Error during get occured: {response.json()}") response.raise_for_status() return response.status_code @@ -231,8 +267,10 @@ def postJson(self, jsonData, url='mo.json') -> {}: # deleteMo # ============================================================================== def deleteMo(self, dn) -> int: - self.__logger.debug(f'Delete Mo called DN: {dn}') - response = self.session.delete(self.baseUrl + "mo/" + dn + ".json", verify=False) + self.__logger.debug(f"Delete Mo called DN: {dn}") + response = self.session.delete( + self.baseUrl + "mo/" + dn + ".json", verify=False + ) # Raise Exception if http Error occurred response.raise_for_status() @@ -243,7 +281,7 @@ def deleteMo(self, dn) -> int: # snapshot # ============================================================================== def snapshot(self, description="snapshot", target_dn="") -> bool: - self.__logger.debug(f'snapshot called {description}') + self.__logger.debug(f"snapshot called {description}") json_payload = [ { @@ -258,7 +296,7 @@ def snapshot(self, description="snapshot", target_dn="") -> bool: "name": "aciclient", "nameAlias": "", "snapshot": "yes", - "targetDn": f"{target_dn}" + "targetDn": f"{target_dn}", } } } @@ -266,13 +304,13 @@ def snapshot(self, description="snapshot", target_dn="") -> bool: response = self.postJson(json_payload) if response == 200: - self.__logger.debug('snapshot created and triggered') + self.__logger.debug("snapshot created and triggered") return True else: - self.__logger.error(f'snapshot creation not succesfull: {response}') + self.__logger.error(f"snapshot creation not succesfull: {response}") return False -# ============================================================================== + # ============================================================================== # subscribe # ============================================================================== def subscribe( diff --git a/aciClient/aciCertClient.py b/aciClient/aciCertClient.py index 27b6ad9..a39c265 100644 --- a/aciClient/aciCertClient.py +++ b/aciClient/aciCertClient.py @@ -24,48 +24,52 @@ class ACICert: # constructor # ============================================================================== def __init__(self, apicIp, pkPath, certDn): - self.__logger.debug(f'Constructor called {apicIp} {pkPath} {certDn}') + self.__logger.debug(f"Constructor called {apicIp} {pkPath} {certDn}") self.apicIp = apicIp - self.baseUrl = 'https://' + self.apicIp + '/api/' - self.__logger.debug(f'BaseUrl set to: {self.baseUrl}') - self.pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, open(pkPath, 'rb').read()) + self.baseUrl = "https://" + self.apicIp + "/api/" + self.__logger.debug(f"BaseUrl set to: {self.baseUrl}") + self.pkey = crypto.load_privatekey( + crypto.FILETYPE_PEM, open(pkPath, "rb").read() + ) self.certDn = certDn # ============================================================================== # packCookies # ============================================================================== def packCookies(self, content) -> {}: - sigResult = base64.b64encode(crypto.sign(self.pkey, content, 'sha256')).decode() + sigResult = base64.b64encode(crypto.sign(self.pkey, content, "sha256")).decode() # print('sigResultType', type(sigResult), sigResult) - return {'APIC-Certificate-Fingerprint': 'fingerprint', - 'APIC-Certificate-Algorithm': 'v1.0', - 'APIC-Certificate-DN': self.certDn, - 'APIC-Request-Signature': sigResult} + return { + "APIC-Certificate-Fingerprint": "fingerprint", + "APIC-Certificate-Algorithm": "v1.0", + "APIC-Certificate-DN": self.certDn, + "APIC-Request-Signature": sigResult, + } # ============================================================================== # getJson # ============================================================================== def getJson(self, uri) -> {}: url = self.baseUrl + uri - self.__logger.debug(f'Get Json called url: {url}') - content = 'GET/api/' + uri + self.__logger.debug(f"Get Json called url: {url}") + content = "GET/api/" + uri cookies = self.packCookies(content) r = requests.get(url, cookies=cookies, verify=False) # Raise Exception if http Error occurred r.raise_for_status() - self.__logger.debug(f'Successful get Data from APIC: {r.json()}') - return r.json()['imdata'] + self.__logger.debug(f"Successful get Data from APIC: {r.json()}") + return r.json()["imdata"] # ============================================================================== # postJson # ============================================================================== def postJson(self, jsonData): - url = self.baseUrl + 'mo.json' - self.__logger.debug(f'Post Json called data: {jsonData}') + url = self.baseUrl + "mo.json" + self.__logger.debug(f"Post Json called data: {jsonData}") data = json.dumps(jsonData, sort_keys=True) - content = 'POST/api/mo.json' + json.dumps(jsonData) + content = "POST/api/mo.json" + json.dumps(jsonData) cookies = self.packCookies(content) r = requests.post(url, data=data, cookies=cookies, verify=False) @@ -73,19 +77,19 @@ def postJson(self, jsonData): r.raise_for_status() if r.status_code == 200: - self.__logger.debug(f'Successful Posted Data to APIC: {r.json()}') + self.__logger.debug(f"Successful Posted Data to APIC: {r.json()}") return r.status_code else: - self.__logger.error(f'Error during get occured: {r.json()}') - return r.status_code, r.json()['imdata'][0]['error']['attributes']['text'] + self.__logger.error(f"Error during get occured: {r.json()}") + return r.status_code, r.json()["imdata"][0]["error"]["attributes"]["text"] # ============================================================================== # deleteMo # ============================================================================== def deleteMo(self, dn): - self.__logger.debug(f'Delete Mo called DN: {dn}') - url = self.baseUrl + 'mo/' + dn + '.json' - content = 'DELETE/api/mo/' + dn + '.json' + self.__logger.debug(f"Delete Mo called DN: {dn}") + url = self.baseUrl + "mo/" + dn + ".json" + content = "DELETE/api/mo/" + dn + ".json" cookies = self.packCookies(content) r = requests.delete(url, cookies=cookies, verify=False) diff --git a/requirements.txt b/requirements.txt index fb966c4..654c5c3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,5 @@ -pyOpenSSL>=23.0.0, <26 -requests[socks]>=2.26.0 , <3 -requests-mock -pytest -flake8 -pysocks==1.7.1 +pyOpenSSL==25.3.0 +requests[socks]==2.32.5 +requests-mock==1.12.1 +pytest==9.0.1 +flake8==7.3.0 diff --git a/setup.py b/setup.py index b46fe63..de3534a 100644 --- a/setup.py +++ b/setup.py @@ -2,20 +2,26 @@ # read the contents of your README file from os import path + this_directory = path.abspath(path.dirname(__file__)) -with open(path.join(this_directory, 'README.md'), encoding='utf-8') as f: +with open(path.join(this_directory, "README.md"), encoding="utf-8") as f: long_description = f.read() -setup(name='aciClient', - version='1.6', - description='aci communication helper class', - url='http://www.netcloud.ch', - author='Netcloud AG', - author_email='nc_dev@netcloud.ch', - license='MIT', - packages=['aciClient'], - install_requires=['requests[socks]>=2.26.0 , <3', 'pyOpenSSL>=23.0.0, <26', 'PySocks>=1.7.1, <2'], - long_description=long_description, - long_description_content_type='text/markdown', - python_requires=">=3.6", - zip_safe=False) +setup( + name="aciClient", + version="1.7", + description="aci communication helper class", + url="http://www.netcloud.ch", + author="Netcloud AG", + author_email="nc_dev@netcloud.ch", + license="MIT", + packages=["aciClient"], + install_requires=[ + "requests[socks]==2.32.5", + "pyOpenSSL==25.3.0", + ], + long_description=long_description, + long_description_content_type="text/markdown", + python_requires=">=3.10", + zip_safe=False, +) diff --git a/test/test_aci.py b/test/test_aci.py index c1ae15e..1be9b75 100644 --- a/test/test_aci.py +++ b/test/test_aci.py @@ -3,280 +3,387 @@ # MIT License # Copyright (c) 2020 Netcloud AG -"""AciClient Testing - -""" +"""AciClient Testing""" from requests import RequestException from aciClient.aci import ACI import pytest import time -__BASE_URL = 'testing-apic.ncdev.ch' +__BASE_URL = "testing-apic.ncdev.ch" def test_login_ok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'asdafa'}}} - ]}) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "asdafa"}}}]}, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") assert aci.login() def test_token_ok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() token = aci.getToken() - assert token == 'tokenxyz' + assert token == "tokenxyz" def test_login_401(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'status_code': 401, 'text': 'not allowed'}, - status_code=401) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"status_code": 401, "text": "not allowed"}, + status_code=401, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") assert not aci.login() def test_login_404_exception(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'status_code': 404, 'text': 'Not Found'}, - status_code=404) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"status_code": 404, "text": "Not Found"}, + status_code=404, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") with pytest.raises(RequestException): resp = aci.login() + def test_login_refresh_ok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'refreshTimeoutSeconds': '31', 'token':'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaRefresh.json', json={ - 'imdata': [ - { - 'aaaLogin': { - 'attributes': { - 'refreshTimeoutSeconds': '300', - 'token':'tokenabc' + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={ + "imdata": [ + { + "aaaLogin": { + "attributes": { + "refreshTimeoutSeconds": "31", + "token": "tokenxyz", + } } } - } - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogout.json', json={'imdata': []}, status_code=200) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown', refresh=True) + ] + }, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaRefresh.json", + json={ + "imdata": [ + { + "aaaLogin": { + "attributes": { + "refreshTimeoutSeconds": "300", + "token": "tokenabc", + } + } + } + ] + }, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogout.json", json={"imdata": []}, status_code=200 + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown", refresh=True) aci.login() token = aci.getToken() time.sleep(2) aci.logout() assert token != aci.getToken() + def test_login_refresh_nok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'refreshTimeoutSeconds': '31', 'token':'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaRefresh.json', json={ - 'imdata': []}, status_code=403) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown', refresh=True) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={ + "imdata": [ + { + "aaaLogin": { + "attributes": { + "refreshTimeoutSeconds": "31", + "token": "tokenxyz", + } + } + } + ] + }, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaRefresh.json", + json={"imdata": []}, + status_code=403, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown", refresh=True) aci.login() time.sleep(3) token = aci.getToken() assert not token + def test_renew_cookie_ok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaRefresh.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}, - status_code=200) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaRefresh.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + status_code=200, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() assert aci.renewCookie() def test_renew_cookie_exception(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaRefresh.json', json={'status_code': 401, 'text': 'Not Allowed'}, - status_code=401) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaRefresh.json", + json={"status_code": 401, "text": "Not Allowed"}, + status_code=401, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() with pytest.raises(RequestException): aci.renewCookie() def test_get_tenant_ok(requests_mock): - uri = 'mo/uni/tn-common.json' - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.get(f'https://{__BASE_URL}/api/{uri}', - json={'imdata': [{'fvTenant': {'attributes': - {'annotation': '', - 'childAction': '', - 'descr': '', - 'dn': 'uni/tn-common', - 'extMngdBy': '', - 'lcOwn': 'local', - 'modTs': '2020-11-23T15:53:52.014+00:00', - 'monPolDn': 'uni/tn-common/monepg-default', - 'name': 'common', - 'nameAlias': '', - 'ownerKey': '', - 'ownerTag': '', - 'status': '', - 'uid': '0'}}}] - }) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + uri = "mo/uni/tn-common.json" + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.get( + f"https://{__BASE_URL}/api/{uri}", + json={ + "imdata": [ + { + "fvTenant": { + "attributes": { + "annotation": "", + "childAction": "", + "descr": "", + "dn": "uni/tn-common", + "extMngdBy": "", + "lcOwn": "local", + "modTs": "2020-11-23T15:53:52.014+00:00", + "monPolDn": "uni/tn-common/monepg-default", + "name": "common", + "nameAlias": "", + "ownerKey": "", + "ownerTag": "", + "status": "", + "uid": "0", + } + } + } + ] + }, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() resp = aci.getJson(uri) - assert 'fvTenant' in resp[0] + assert "fvTenant" in resp[0] def test_get_tenant_ok(requests_mock): - uri = 'mo/uni/tn-common.json' - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.get(f'https://{__BASE_URL}/api/{uri}', - json={'imdata': [{'fvTenant': {'attributes': - {'annotation': '', - 'childAction': '', - 'descr': '', - 'dn': 'uni/tn-common', - 'extMngdBy': '', - 'lcOwn': 'local', - 'modTs': '2020-11-23T15:53:52.014+00:00', - 'monPolDn': 'uni/tn-common/monepg-default', - 'name': 'common', - 'nameAlias': '', - 'ownerKey': '', - 'ownerTag': '', - 'status': '', - 'uid': '0'}}}] - }) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + uri = "mo/uni/tn-common.json" + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.get( + f"https://{__BASE_URL}/api/{uri}", + json={ + "imdata": [ + { + "fvTenant": { + "attributes": { + "annotation": "", + "childAction": "", + "descr": "", + "dn": "uni/tn-common", + "extMngdBy": "", + "lcOwn": "local", + "modTs": "2020-11-23T15:53:52.014+00:00", + "monPolDn": "uni/tn-common/monepg-default", + "name": "common", + "nameAlias": "", + "ownerKey": "", + "ownerTag": "", + "status": "", + "uid": "0", + } + } + } + ] + }, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() resp = aci.getJson(uri) - assert 'fvTenant' in resp[0] + assert "fvTenant" in resp[0] def test_get_tenant_not_found(requests_mock): - uri = 'mo/uni/tn-commmmmon.json' - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.get(f'https://{__BASE_URL}/api/{uri}', - json={'error': 'Not found'}, status_code=404) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + uri = "mo/uni/tn-commmmmon.json" + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.get( + f"https://{__BASE_URL}/api/{uri}", json={"error": "Not found"}, status_code=404 + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() resp = aci.getJson(uri) - assert 'error' in resp + assert "error" in resp def test_post_tenant_ok(requests_mock): - post_data = {'fvTenant': {'attributes': - {'descr': '', - 'dn': 'uni/tn-test', - 'status': 'modified,created'}} - } - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', - json={'imdata': post_data}) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + post_data = { + "fvTenant": { + "attributes": { + "descr": "", + "dn": "uni/tn-test", + "status": "modified,created", + } + } + } + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.post(f"https://{__BASE_URL}/api/mo.json", json={"imdata": post_data}) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() resp = aci.postJson(post_data) assert resp == 200 def test_post_tenant_bad_request(requests_mock): - post_data = {'fvTenFailFailant': {'attributes': - {'descr': '', - 'dn': 'uni/tn-test', - 'status': 'modified,created'}} - } - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', - json={'imdata': [ - {'error': {'attributes': {'text': 'tokenxyz'}}}]}, - status_code=400) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + post_data = { + "fvTenFailFailant": { + "attributes": { + "descr": "", + "dn": "uni/tn-test", + "status": "modified,created", + } + } + } + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/mo.json", + json={"imdata": [{"error": {"attributes": {"text": "tokenxyz"}}}]}, + status_code=400, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() resp = aci.postJson(post_data) - assert resp.startswith('400: ') + assert resp.startswith("400: ") def test_post_tenant_forbidden_exception(requests_mock): - post_data = {'fvTenFailFailant': {'attributes': - {'descr': '', - 'dn': 'uni/tn-test', - 'status': 'modified,created'}} - } - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', - json={'imdata': [ - {'error': {'attributes': {'text': 'tokenxyz'}}}]}, - status_code=403) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + post_data = { + "fvTenFailFailant": { + "attributes": { + "descr": "", + "dn": "uni/tn-test", + "status": "modified,created", + } + } + } + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/mo.json", + json={"imdata": [{"error": {"attributes": {"text": "tokenxyz"}}}]}, + status_code=403, + ) + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() with pytest.raises(RequestException): aci.postJson(post_data) def test_snapshot_ok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', json={"totalCount": "0", "imdata": []}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/mo.json", json={"totalCount": "0", "imdata": []} + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() - resp = aci.snapshot(description='unit_test') + resp = aci.snapshot(description="unit_test") assert resp def test_snapshot_nok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', - json={"totalCount": "0", "imdata": [{"error": {"attributes": {"text": "Error UnitTest"}}}]}, - status_code=400) - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/mo.json", + json={ + "totalCount": "0", + "imdata": [{"error": {"attributes": {"text": "Error UnitTest"}}}], + }, + status_code=400, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() - resp = aci.snapshot(description='unit_test') + resp = aci.snapshot(description="unit_test") assert not resp -def test_snapshot_tenant_ok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', json={"totalCount": "0", "imdata": []}) - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') +def test_snapshot_tenant_ok(requests_mock): + requests_mock.post( + f"https://{__BASE_URL}/api/mo.json", json={"totalCount": "0", "imdata": []} + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() - resp = aci.snapshot(description='unit_test', target_dn='/uni/tn-test') + resp = aci.snapshot(description="unit_test", target_dn="/uni/tn-test") assert resp def test_snapshot_tenant_nok(requests_mock): - requests_mock.post(f'https://{__BASE_URL}/api/mo.json', - json={"totalCount": "0", "imdata": [{"error": {"attributes": {"text": "Error UnitTest"}}}]}, - status_code=400) - requests_mock.post(f'https://{__BASE_URL}/api/aaaLogin.json', json={'imdata': [ - {'aaaLogin': {'attributes': {'token': 'tokenxyz'}}} - ]}) - - aci = ACI(apicIp=__BASE_URL, apicUser='admin', apicPasword='unkown') + requests_mock.post( + f"https://{__BASE_URL}/api/mo.json", + json={ + "totalCount": "0", + "imdata": [{"error": {"attributes": {"text": "Error UnitTest"}}}], + }, + status_code=400, + ) + requests_mock.post( + f"https://{__BASE_URL}/api/aaaLogin.json", + json={"imdata": [{"aaaLogin": {"attributes": {"token": "tokenxyz"}}}]}, + ) + + aci = ACI(apicIp=__BASE_URL, apicUser="admin", apicPasword="unkown") aci.login() - resp = aci.snapshot(description='unit_test', target_dn='/uni/tn-test') + resp = aci.snapshot(description="unit_test", target_dn="/uni/tn-test") assert not resp