✅ added test for source, task and flow

src/viadot/sources/business_core.py

@@ -26,7 +26,7 @@ class BusinessCoreCredentials(BaseModel):
 
 class BusinessCore(Source):
     """
-    Source for getting data from Bussines Core ERP API.
+    Source for getting data from Business Core ERP API.
     """
     def __init__(
         self,

tests/integration/orchestration/prefect/flows/test_business_core_to_parquet.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+from pandas import DataFrame, read_parquet
+from viadot.orchestration.prefect.flows import business_core_to_parquet
+
+URL="https://api.businesscore.ae/api/LappDataIntegrationAPI/GetCustomerData"
+PATH="/home/viadot/data/middle_east/customer_master/customer_master_full_data.parquet"
+CREDS = "business-core"
+
+
+def test_business_core_to_parquet():
+    assert not Path(PATH).exists()
+
+    business_core_to_parquet(
+        url=URL, 
+        path=PATH, 
+        credentials_secret=CREDS,
+        verify=False
+        )
+
+    assert Path(PATH).exists()
+
+    n_cols = 11
+
+    df = read_parquet(PATH)
+
+    assert df.shape[1] == n_cols
+    Path(PATH).unlink()

tests/integration/orchestration/prefect/tasks/test_business_core.py

@@ -0,0 +1,29 @@
+from pathlib import Path
+from pandas import DataFrame
+
+from prefect import flow
+from viadot.orchestration.prefect.tasks.business_core import business_core_to_df
+
+URL="https://api.businesscore.ae/api/LappDataIntegrationAPI/GetCustomerData"
+PATH="/home/viadot/data/middle_east/customer_master/customer_master_full_data.parquet"
+CREDS = "business-core"
+
+
+def test_business_core_to_df():
+    @flow
+    def test_flow():
+        return business_core_to_df(
+            url=URL, 
+            path=PATH, 
+            credentials_secret=CREDS,
+            verify=False
+            )
+
+    df = test_flow()
+    assert isinstance(df, DataFrame)
+
+    n_cols = 11
+    assert df.shape[1] == n_cols
+
+
+

tests/integration/test_business_core.py

@@ -0,0 +1,50 @@
+from unittest.mock import Mock, patch
+import pandas as pd
+import pytest
+
+from viadot.sources.business_core import BusinessCore
+
+
+@pytest.fixture(scope="module")
+def business_core():
+    return BusinessCore(
+        url="https://api.businesscore.ae/api/GetCustomerData",
+        filters_dict={
+            "BucketCount": 10,
+            "BucketNo": 1,
+            "FromDate": None,
+            "ToDate": None,
+        },
+        credentials={"username": "test", "password": "test123"},
+    )
+
+
+@patch("viadot.sources.business_core.handle_api_response")
+def test_generate_token(mock_api_response, business_core):
+    mock_api_response.return_value = Mock(text='{"access_token": "12345"}')
+    token = business_core.generate_token()
+    assert token == "12345"
+
+
+def test_clean_filters_dict(business_core):
+    filters = business_core.clean_filters_dict()
+    assert filters == {
+        "BucketCount": 10,
+        "BucketNo": 1,
+        "FromDate": "&",
+        "ToDate": "&",
+    }
+
+
+def test_to_df(business_core):
+    with patch.object(
+        business_core,
+        "get_data",
+        return_value={"MasterDataList": [{"id": 1, "name": "John Doe"}]},
+    ):
+        df = business_core.to_df()
+        assert isinstance(df, pd.DataFrame)
+        assert len(df.columns) == 2
+        assert len(df) == 1
+        assert df["id"].tolist() == [1]
+        assert df["name"].tolist() == ["John Doe"]