+6Hi Team,
I need help to write the unit test using pytest,
below is the code:
row_list = cognite.raw.rows.list(file.destination.database, file.Extractor_Last_Run_Table, limit=-1,
columns=["key", "LastUpdatedDatetime"]).to_pandas()
if not row_list.empty:
val = list(row_list[row_list.index.isin([file.Extractor_key])]['LastUpdatedDatetime'])
Could you please help to write the unit test for above code?
Thanks
Best answer by Nimesha
Hi Harshita,
Sure, please adjusted the code as shown below;
Check the
documentation
Ask the
Community
Take a look
at
Academy
Cognite
Status
Page
Contact
Cognite Support
import pytest
from unittest.mock import MagicMock, patch
import pandas as pd
from cognite.client import CogniteClient
c = CogniteClient()
# Assuming you have a function process_data in your code
def process_data(database, table, key):
row_list = c.raw.rows.list(database, table, limit=-1, columns=["key", "LastUpdatedDatetime"]).to_pandas()
if not row_list.empty:
val = list(row_list[row_list.index.isin([key])]['LastUpdatedDatetime'])
return val
else:
return None
# Assuming you have a CogniteClient class defined in your code
class CogniteClient:
def __init__(self):
self.raw = MagicMock()
# Your test function
@pytest.mark.asyncio
async def test_your_function():
# Set up your test data
database = "your_database"
table = "your_table"
key = "your_key"
# Create a CogniteClient instance
c = CogniteClient()
# Set up the mock response for c.raw.rows.list
last_updated_datetime = "2023-03-16T17:51:54Z"
# Create a MagicMock for the list
mock_list = MagicMock()
# Set up the mock response for the list
mock_list.return_value = [{"key": "your_key", "LastUpdatedDatetime": last_updated_datetime}]
# Set the return value of to_pandas on the list
mock_list.to_pandas.return_value = pd.DataFrame(mock_list.return_value)
# Use patch to temporarily replace the method with the mock_list
with patch.object(c.raw.rows, 'list', return_value=mock_list):
# Call the function
result = process_data(database, table, key)
# Assert the expected behavior
assert result == ["2023-03-16T17:51:54Z"]I have tested the above code with our test environment and able to passed the unit mock test. Please change the LastUpdatedDatetime, database, table, key with your environment values.
plugins: asyncio-0.21.0, cov-3.0.0, rerunfailures-11.1.2, xdist-3.2.1
asyncio: mode=strict
collected 1 item
test_data_processor.py . [100%]
============================================================================================================ 1 passed in 12.33s ============================================================================================================
Hi Harshita,
Thank you for the questions.
Let's create a simple example. Assume you have a module named data_processor.py with the following function:
from cognite.client import CogniteClient
c = CogniteClient()
def process_data(database, table, key):
row_list = c.raw.rows.list(database, table, limit=-1, columns=["key", "LastUpdatedDatetime"]).to_pandas()
if not row_list.empty:
val = list(row_list[row_list.index.isin([key])]['LastUpdatedDatetime'])
return val
else:
return NoneNow, you want to write a test for this function using pytest. Create a file named test_data_processor.py:
import pytest
from data_processor import process_data
import pandas as pd
from unittest.mock import patch
from unittest.mock import MagicMock
from cognite.client import CogniteClient
c = CogniteClient()
@pytest.fixture
def mock_cognite_raw_rows_list():
c = MagicMock()
mock_response_data = {
"key": ["test_key"],
"LastUpdatedDatetime": ["2023-03-16T17:51:54Z"]
}
c.raw.rows.list.return_value.to_pandas.return_value = pd.DataFrame(mock_response_data)
# Call the mock function
row_list = c.raw.rows.list("test_db", "test_table", limit=5)
return row_list
def test_process_data(mock_cognite_raw_rows_list):
# Set up your test data
database = "test_db"
table = "test_table"
key = "test_key"
# Set up mock response from cognite.raw.rows.list
mock_response_data = {
"key": [key],
"LastUpdatedDatetime": ["2023-03-16T17:51:54Z"]
}
mock_cognite_raw_rows_list.return_value.to_pandas.return_value = pd.DataFrame(mock_response_data)
# Call your function
result = process_data(database, table, key)
# Assert the expected behavior
assert result == ["2023-03-16T17:51:54Z"]
mock_cognite_raw_rows_list.assert_called_once_with(
database,
table,
limit=-1,
columns=["key", "LastUpdatedDatetime"]
)
To run the test, open a terminal and run:
pytest test_data_processor.py
Make sure to install pytest and pytest-mock first:
pip install pytest pytest-mockMake sure you have the cognite-sdk package installed:
pip install cognite-sdkPlease fine the document for the List rows in a table here.
+6Hi Nimesha,
Thanks a lot for the code. I have tried the above code.
I have small doubt. When I generate the coverage report the it’s showing the below line is miss, can you help me how the list will be mock?
if not row_list.empty:
val = list(row_list[row_list.index.isin([key])]['LastUpdatedDatetime'])
Regards,
Sai Harshita
Hi Harshita,
Sure, please adjusted the code as shown below;
import pytest
from unittest.mock import MagicMock, patch
import pandas as pd
from cognite.client import CogniteClient
c = CogniteClient()
# Assuming you have a function process_data in your code
def process_data(database, table, key):
row_list = c.raw.rows.list(database, table, limit=-1, columns=["key", "LastUpdatedDatetime"]).to_pandas()
if not row_list.empty:
val = list(row_list[row_list.index.isin([key])]['LastUpdatedDatetime'])
return val
else:
return None
# Assuming you have a CogniteClient class defined in your code
class CogniteClient:
def __init__(self):
self.raw = MagicMock()
# Your test function
@pytest.mark.asyncio
async def test_your_function():
# Set up your test data
database = "your_database"
table = "your_table"
key = "your_key"
# Create a CogniteClient instance
c = CogniteClient()
# Set up the mock response for c.raw.rows.list
last_updated_datetime = "2023-03-16T17:51:54Z"
# Create a MagicMock for the list
mock_list = MagicMock()
# Set up the mock response for the list
mock_list.return_value = [{"key": "your_key", "LastUpdatedDatetime": last_updated_datetime}]
# Set the return value of to_pandas on the list
mock_list.to_pandas.return_value = pd.DataFrame(mock_list.return_value)
# Use patch to temporarily replace the method with the mock_list
with patch.object(c.raw.rows, 'list', return_value=mock_list):
# Call the function
result = process_data(database, table, key)
# Assert the expected behavior
assert result == ["2023-03-16T17:51:54Z"]I have tested the above code with our test environment and able to passed the unit mock test. Please change the LastUpdatedDatetime, database, table, key with your environment values.
plugins: asyncio-0.21.0, cov-3.0.0, rerunfailures-11.1.2, xdist-3.2.1
asyncio: mode=strict
collected 1 item
test_data_processor.py . [100%]
============================================================================================================ 1 passed in 12.33s ============================================================================================================
Enter your E-mail address. We'll send you an e-mail with instructions to reset your password.