diff --git a/.github/workflows/mr_ci.yml b/.github/workflows/mr_ci.yml index 40c29b3..2cd48ab 100644 --- a/.github/workflows/mr_ci.yml +++ b/.github/workflows/mr_ci.yml @@ -34,8 +34,8 @@ jobs: - name: Lint Step run: | black --check --target-version py310 . - isort --check --profile black --skip env . - flake8 --count --show-source --statistics --exclude .git,env,__init__.py --max-line-length 88 . + isort --check --profile black --skip build_env --skip test_env . + flake8 --count --show-source --statistics --exclude .git,build_env,test_env,__init__.py --max-line-length 88 . - name: Coverage with pytest run: | pytest --no-cov-on-fail --cov argocd --cov-report html --cov-report term --cov-report xml tests/unit diff --git a/argocd/services/session.py b/argocd/services/session.py new file mode 100644 index 0000000..8e40239 --- /dev/null +++ b/argocd/services/session.py @@ -0,0 +1,37 @@ +from requests import sessions + +from argocd import config + + +class SessionService: + def __init__(self): + self.config = config.Config() + self.session = sessions.Session() + self.base_url = self.config.server_url + + def create_session(self, password, token, username): + """ + Create a new JWT for authentication and set a cookie if using HTTP. + """ + response = self.session.post( + f"{self.base_url}/api/v1/session", + json={"password": password, "token": token, "username": username}, + ) + + return response.json() + + def delete_session(self): + """ + Delete an existing JWT cookie if using HTTP + """ + response = self.session.delete(f"{self.base_url}/api/v1/session") + + return response.json() + + def me(self): + """ + Get the current user's info + """ + response = self.session.get(f"{self.base_url}/api/v1/session/userinfo") + + return response.json() diff --git a/tests/unit/services/test_session.py b/tests/unit/services/test_session.py new file mode 100644 index 0000000..7487d52 --- /dev/null +++ b/tests/unit/services/test_session.py @@ -0,0 +1,83 @@ +import os +import unittest.mock + +import pytest + +from argocd.services import session +from tests.services import mocks + + +@pytest.fixture +def session_service(): + os.environ["ARGOCD_API_BASE_URL"] = "http://demo.com" + return session.SessionService() + + +@unittest.mock.patch( + "argocd.services.session.sessions.Session.get", + mocks.mocked_success_session_me_response, +) +def test_session_api_me_returns_ok(session_service): + response = session_service.me() + + assert isinstance(response, dict) + assert "username" in response + assert "groups" in response + assert isinstance(response["groups"], list) + + +@unittest.mock.patch( + "argocd.services.session.sessions.Session.get", + mocks.mocked_failure_generic_response, +) +def test_session_api_me_returns_non_ok(session_service): + response = session_service.me() + + assert isinstance(response, dict) + assert "error" in response.keys() + + +@unittest.mock.patch( + "argocd.services.session.sessions.Session.post", + mocks.mocked_success_session_create_and_delete_session_response, +) +def test_session_api_create_session_returns_ok(session_service): + response = session_service.create_session("password", "token", "username") + + assert isinstance(response, dict) + assert "token" in response.keys() + assert isinstance(response["token"], str) + + +@unittest.mock.patch( + "argocd.services.session.sessions.Session.post", + mocks.mocked_failure_generic_response, +) +def test_session_api_create_session_returns_not_ok(session_service): + response = session_service.create_session("password", "token", "username") + + assert isinstance(response, dict) + assert "error" in response.keys() + + +@unittest.mock.patch( + "argocd.services.session.sessions.Session.delete", + mocks.mocked_success_session_create_and_delete_session_response, +) +def test_session_api_delete_session_returns_ok(session_service): + response = session_service.delete_session() + + assert isinstance(response, dict) + assert "token" in response.keys() + assert isinstance(response["token"], str) + + +@unittest.mock.patch( + "argocd.services.session.sessions.Session.delete", + mocks.mocked_failure_generic_response, +) +def test_session_api_delete_session_returns_not_ok(session_service): + response = session_service.delete_session() + + assert isinstance(response, dict) + assert "error" in response.keys()