From: Swaraj Kumar <swaraj.kumar@samsung.com>
Date: Mon, 13 Jan 2025 19:49:16 +0000 (+0530)
Subject: UT for pipeline manager
X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=commitdiff_plain;h=refs%2Fchanges%2F06%2F14006%2F1;p=aiml-fw%2Fawmf%2Ftm.git

UT for pipeline manager

Change-Id: I6ed5122d3747c21b0be0f019be8c4e2c9f318e11
Signed-off-by: Swaraj Kumar <swaraj.kumar@samsung.com>
---

diff --git a/tests/test_pipelinemgr.py b/tests/test_pipelinemgr.py
new file mode 100644
index 0000000..92d972d
--- /dev/null
+++ b/tests/test_pipelinemgr.py
@@ -0,0 +1,52 @@
+import pytest
+from flask import Flask
+import json
+import requests
+from unittest import mock
+from unittest.mock import patch, MagicMock
+import pytest
+from requests.models import Response
+from threading import Lock
+import os
+import sys
+import datetime
+from flask_api import status
+from dotenv import load_dotenv
+load_dotenv('tests/test.env')
+from trainingmgr.constants.states import States
+from threading import Lock
+from trainingmgr.common.tmgr_logger import TMLogger
+from trainingmgr.common.trainingmgr_config import TrainingMgrConfig
+from trainingmgr.common.exceptions_utls import DBException, TMException
+from trainingmgr.models import TrainingJob
+from trainingmgr.models import FeatureGroup
+from trainingmgr.common.trainingConfig_parser import getField
+from unittest.mock import patch, MagicMock
+from trainingmgr.pipeline.pipeline_mgr import PipelineMgr
+from trainingmgr.common.exceptions_utls import TMException
+
+@pytest.fixture
+def pipeline_mgr():
+    """Fixture to get an instance of PipelineMgr."""
+    return PipelineMgr()
+    
+class TestPipelineMgr:
+    @patch("requests.get")
+    def test_get_all_pipelines_success(self, mock_get, pipeline_mgr):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.headers = {"content-type": "application/json"}
+        mock_response.json.return_value = [{"id": "pipeline1"}, {"id": "pipeline2"}]
+        mock_get.return_value = mock_response
+        result = pipeline_mgr.get_all_pipelines()
+        assert len(result) == 2
+        assert result[0]["id"] == "pipeline1"
+        
+    @patch("requests.get")
+    def test_get_all_pipelines_invalid_response(self, mock_get, pipeline_mgr):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.headers = {"content-type": "text/plain"}
+        mock_get.return_value = mock_response
+        with pytest.raises(TMException, match="Kf adapter doesn't sends json type response"):
+            pipeline_mgr.get_all_pipelines()