added tests

.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
+          flake8 --ignore=E501,W293 vue-python-template/python/src vue-python-template/tests --show-source

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: <class 'ValueError'> 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: <class 'ValueError'> Headers must be a dictionary")

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

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'))

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()