Adds Session Service

.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

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

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