From 76fcf2797451563d8da7b04703b1f8c49785192c Mon Sep 17 00:00:00 2001 From: nk-randers Date: Tue, 22 Oct 2024 09:20:46 +0200 Subject: [PATCH] added tests --- .github/workflows/lint.yaml | 4 +- tests/api_request_test.py | 181 ++++++++++++++++++++++++++++++++++++ tests/conf_test.py | 30 ++++++ tests/logging_test.py | 43 +++++++++ tests/sftp_test.py | 37 ++++++++ 5 files changed, 293 insertions(+), 2 deletions(-) create mode 100644 tests/api_request_test.py create mode 100644 tests/conf_test.py create mode 100644 tests/logging_test.py create mode 100644 tests/sftp_test.py diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml index 012ed44..726d802 100644 --- a/.github/workflows/lint.yaml +++ b/.github/workflows/lint.yaml @@ -32,7 +32,7 @@ jobs: pip install -r requirements-dev.txt - name: Check for syntax errors or undefined names run: | - flake8 src tests --count --select=E9,F63,F7,F82 --show-source --statistics + flake8 vue-python-template/python/src vue-python-template/tests --count --select=E9,F63,F7,F82 --show-source --statistics - name: Lint with flake8 run: | - flake8 --ignore=E501,W293 src tests --show-source \ No newline at end of file + flake8 --ignore=E501,W293 vue-python-template/python/src vue-python-template/tests --show-source \ No newline at end of file diff --git a/tests/api_request_test.py b/tests/api_request_test.py new file mode 100644 index 0000000..5b30fb1 --- /dev/null +++ b/tests/api_request_test.py @@ -0,0 +1,181 @@ +import pytest +import base64 + +from unittest.mock import MagicMock, patch + +from utils.api_requests import APIClient + + +def test_init(): + api_client = APIClient('http://testurl.com') + + assert api_client.base_url == 'http://testurl.com' + +# _authenticate tests + + +def test_authenticate_api_key(): + api_client = APIClient('http://testurl.com', api_key='test_key') + + assert api_client._authenticate() == {'Authorization': 'Bearer test_key'} + + +@patch('time.time') +@patch('requests.post') +def test_authenticate_client_credentials(mock_post, mock_time): + api_client = APIClient('http://testurl.com', client_id='test_id', client_secret='test_secret', realm='test_realm') + + mock_time.return_value = 0 + + res = MagicMock() + res.raise_for_status.return_value = None + res.json.return_value = {'access_token': 'test_token', 'expires_in': 10, 'refresh_token': 'test_refresh_token', 'refresh_expires_in': 20} + mock_post.return_value = res + + assert api_client._authenticate() == {'Authorization': 'Bearer test_token'} + mock_post.assert_called_once_with('http://testurl.com/auth/realms/test_realm/protocol/openid-connect/token', headers={'Content-Type': 'application/x-www-form-urlencoded'}, data={'client_id': 'test_id', 'client_secret': 'test_secret', 'grant_type': 'client_credentials'}) + assert api_client.token_expiry == 10 + assert api_client.refresh_token_expiry == 20 + assert api_client.refresh_token == 'test_refresh_token' + + assert api_client._authenticate() == {'Authorization': 'Bearer test_token'} + + +@patch('time.time') +@patch('requests.post') +def test_authenticate_user_password(mock_post, mock_time): + api_client = APIClient('http://testurl.com', client_id='test_id', client_secret='test_secret', realm='test_realm', username='test_user', password='test_pass') + + mock_time.return_value = 0 + + res = MagicMock() + res.raise_for_status.return_value = None + res.json.return_value = {'access_token': 'test_token', 'expires_in': 10, 'refresh_token': 'test_refresh_token', 'refresh_expires_in': 20} + mock_post.return_value = res + + assert api_client._authenticate() == {'Authorization': 'Bearer test_token'} + mock_post.assert_called_once_with('http://testurl.com/auth/realms/test_realm/protocol/openid-connect/token', headers={'Content-Type': 'application/x-www-form-urlencoded'}, data={'client_id': 'test_id', 'client_secret': 'test_secret', 'grant_type': 'password', 'username': 'test_user', 'password': 'test_pass'}) + assert api_client.token_expiry == 10 + assert api_client.refresh_token_expiry == 20 + assert api_client.refresh_token == 'test_refresh_token' + + +def test_authenticate_cert(): + test_base64 = base64.b64encode(b'test_cert') + api_client = APIClient('http://testurl.com', cert_base64=test_base64, password='test_pass') + + assert api_client._authenticate() == {} + + +@patch('time.time') +@patch('requests.post') +def test_authenticate_refresh_token(mock_post, mock_time): + api_client = APIClient('http://testurl.com', client_id='test_id', client_secret='test_secret', realm='test_realm') + api_client.access_token = 'test_token' + api_client.token_expiry = 10 + api_client.refresh_token = 'test_refresh_token' + api_client.refresh_token_expiry = 20 + + mock_time.return_value = 15 + + res = MagicMock() + res.raise_for_status.return_value = None + res.json.return_value = {'access_token': 'test_token', 'expires_in': 10, 'refresh_token': 'test_refresh_token', 'refresh_expires_in': 20} + mock_post.return_value = res + + assert api_client._authenticate() == {'Authorization': 'Bearer test_token'} + mock_post.assert_called_once_with('http://testurl.com/auth/realms/test_realm/protocol/openid-connect/token', headers={'Content-Type': 'application/x-www-form-urlencoded'}, data={'client_id': 'test_id', 'client_secret': 'test_secret', 'grant_type': 'refresh_token', 'refresh_token': 'test_refresh_token'}) + assert api_client.token_expiry == 25 # time now is 15 + assert api_client.refresh_token_expiry == 35 # time now is 15 + assert api_client.refresh_token == 'test_refresh_token' + + +def test_authenticate_no_realm(): + with pytest.raises(ValueError) as excinfo: + api_client = APIClient('http://testurl.com', client_id='test_id', client_secret='test_secret', realm='test_realm') + api_client.realm = None + api_client._authenticate() + assert 'Realm is required for client_id and client_secret authentication' in str(excinfo.value) + +# make_request tests + + +@patch('requests.get') +def test_make_request_get(mock_get): + api_client = APIClient('http://testurl.com', api_key='test_key') + + res = MagicMock() + res.raise_for_status.return_value = None + res.headers.get.return_value = 'application/json' + res.json.return_value = {'test': 'test'} + mock_get.return_value = res + + assert api_client.make_request(path='/test') == {'test': 'test'} + mock_get.assert_called_once_with('http://testurl.com/test', headers={'Authorization': 'Bearer test_key'}) + + +@patch('requests.post') +def test_make_request_post(mock_get): + api_client = APIClient('http://testurl.com', api_key='test_key') + + res = MagicMock() + res.raise_for_status.return_value = None + res.content = b'' + mock_get.return_value = res + + assert api_client.make_request(path='/test', json={'test': 'test'}) == b' ' + mock_get.assert_called_once_with('http://testurl.com/test', headers={'Authorization': 'Bearer test_key', 'Content-Type': 'application/json'}, json={'test': 'test'}) + + +@patch('requests.put') +def test_make_request_put(mock_get): + api_client = APIClient('http://testurl.com', api_key='test_key') + + res = MagicMock() + res.raise_for_status.return_value = None + res.content = b'ok' + mock_get.return_value = res + + assert api_client.make_request(method='put', headers={'custom': 'header'}, data='test') == b'ok' + mock_get.assert_called_once_with('http://testurl.com', headers={'Authorization': 'Bearer test_key', 'custom': 'header'}, data='test') + + +@patch('requests.delete') +def test_make_request_delete(mock_get): + api_client = APIClient('http://testurl.com', api_key='test_key') + + res = MagicMock() + res.raise_for_status.return_value = None + res.content = b'ok' + mock_get.return_value = res + + assert api_client.make_request(method='DELETE', path='/test', data='test') == b'ok' + mock_get.assert_called_once_with('http://testurl.com/test', headers={'Authorization': 'Bearer test_key'}, data='test') + + +@patch('requests_pkcs12.post') +def test_make_request_get_cert(mock_get): + test_base64 = base64.b64encode(b'test_cert') + api_client = APIClient('http://testurl.com', cert_base64=test_base64, password='test_pass') + + res = MagicMock() + res.raise_for_status.return_value = None + res.content = b'ok' + mock_get.return_value = res + + assert api_client.make_request(path='/test', json={'test': 'test'}) == b'ok' + mock_get.assert_called_once_with('http://testurl.com/test', json={'test': 'test'}, pkcs12_data=b'test_cert', pkcs12_password='test_pass', headers={'Content-Type': 'application/json'}) + + +def test_make_request_wrong_path(): + api_client = APIClient('http://testurl.com', api_key='test_key') + api_client.logger = MagicMock() + assert api_client.make_request(path=1) is None + api_client.logger.error.assert_called_once_with("Request failed with error: Path must be a string") + + +def test_make_request_wrong_headers(): + api_client = APIClient('http://testurl.com', api_key='test_key') + api_client.logger = MagicMock() + assert api_client.make_request(headers='not a dict') is None + api_client.logger.error.assert_called_once_with("Request failed with error: Headers must be a dictionary") diff --git a/tests/conf_test.py b/tests/conf_test.py new file mode 100644 index 0000000..e426568 --- /dev/null +++ b/tests/conf_test.py @@ -0,0 +1,30 @@ +import pytest +from main import create_app + + +@pytest.fixture() +def app(): + app = create_app() + app.config.update({ + "TESTING": True, + }) + + yield app + + +@pytest.fixture() +def client(app): + return app.test_client() + + +def test_healthz(client): + response = client.get('/healthz') + assert response.status_code == 200 + assert response.get_json()['status'] == 'success' + + +def test_metrics(client): + # POD_NAME env var set in pytest.ini (test-pod) + response = client.get('/metrics') + assert response.status_code == 200 + assert 'is_ready gauge\nis_ready{error_type="None",job_name="test-pod"} 1.0' in response.text diff --git a/tests/logging_test.py b/tests/logging_test.py new file mode 100644 index 0000000..c39e031 --- /dev/null +++ b/tests/logging_test.py @@ -0,0 +1,43 @@ +import logging +import sys + +from unittest.mock import patch, Mock +from werkzeug import serving + +from utils.logging import disable_endpoint_logs, set_logging_configuration + + +def test_disable_endpoint_logs(): + original_log_request = Mock() + with patch.object(serving.WSGIRequestHandler, 'log_request', new=original_log_request): + + disable_endpoint_logs(['/disabled-endpoint']) + + mock_request = Mock() + mock_request.path = '/disabled-endpoint' + + serving.WSGIRequestHandler.log_request(mock_request) + + original_log_request.assert_not_called() + + mock_request.path = '/other' + + serving.WSGIRequestHandler.log_request(mock_request) + + original_log_request.assert_called_once_with(mock_request) + + +@patch('logging.basicConfig') +@patch('utils.logging.disable_endpoint_logs') +def test_set_logging_configuration(mock_disable_endpoint_logs, mock_basicConfig): + + set_logging_configuration() + + mock_basicConfig.assert_called_once_with( + stream=sys.stdout, + level=logging.INFO, + format='[%(asctime)s] %(levelname)s - %(name)s - %(module)s:%(funcName)s - %(message)s', + datefmt='%d-%m-%Y %H:%M:%S' + ) + + mock_disable_endpoint_logs.assert_called_once_with(('/metrics', '/healthz')) diff --git a/tests/sftp_test.py b/tests/sftp_test.py new file mode 100644 index 0000000..a40d643 --- /dev/null +++ b/tests/sftp_test.py @@ -0,0 +1,37 @@ +from utils.sftp import SFTPClient +from unittest.mock import patch, MagicMock + + +@patch('pysftp.Connection') +def test_get_connection_success(mock_connection): + client = SFTPClient('host', 'user', 'pass') + mock_connection.return_value = MagicMock() + + result = client.get_connection() + + assert result is not None + mock_connection.assert_called_once_with(host='host', username='user', password='pass', private_key=None, cnopts=client.cnopts) + + +@patch('pysftp.Connection') +def test_get_connection_exception(mock_connection): + + client = SFTPClient('host', 'user', 'pass') + mock_connection.side_effect = Exception("Connection failed") + + result = client.get_connection() + + assert result is None + + +@patch('base64.b64decode') +@patch('paramiko.RSAKey.from_private_key') +def test_make_key(mock_from_private_key, mock_b64decode): + mock_b64decode.return_value = b'decoded_key' + mock_from_private_key.return_value = 'RSAKey' + + client = SFTPClient('host', 'user', key_base64='base64key', key_pass='keypass') + + assert client.key == 'RSAKey' + mock_b64decode.assert_called_once_with('base64key') + mock_from_private_key.assert_called_once()