Solved

How to write the unit test for the list row?

  • 1 November 2023
  • 4 replies
  • 97 views
H
Rank
Badge +6

Hi 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

icon

Best answer by Nimesha 3 November 2023, 09:34

View original

4 replies

N
Rank

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  None

Now, 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-mock

Make sure you have the cognite-sdk package installed:

pip install cognite-sdk

Please fine the document for the List rows in a table  here. 

H
Rank
Badge +6

Hi 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

N
Rank

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 ============================================================================================================

 

H
Rank
Badge +6

Thank you Nimesha it’s work for me.😀

Reply


Terms & ConditionsCookie Settings