Testing
Koza includes a mock_koza
fixture (see src/koza/utils/testing_utils
) that can be used to test your ingest configuration. This fixture accepts the following arguments:
Argument | Type | Description |
---|---|---|
Required Arguments | ||
name |
str |
The name of the ingest |
data |
Union[Dict, List[Dict]] |
The data to be ingested |
transform_code |
str |
Path to the transform code to be used |
Optional Arguments | ||
map_cache |
Dict |
Map cache to be used |
filters |
List(str) |
List of filters to apply to data |
global_table |
str |
Path to the global table |
local_table |
str |
Path to the local table |
The mock_koza
fixture returns a list of entities that would be generated by the ingest configuration.
This list can then be used to test the output based on the transform script.
Here is an example of how to use the mock_koza
fixture to test an ingest configuration:
import pytest
from koza.utils.testing_utils import mock_koza
# Define the source name and transform script path
INGEST_NAME = "your_ingest_name"
TRANSFORM_SCRIPT = "./src/{{cookiecutter.__project_slug}}/transform.py"
# Define an example row to test (as a dictionary)
@pytest.fixture
def example_row():
return {
"example_column_1": "entity_1",
"example_column_2": "entity_6",
"example_column_3": "biolink:related_to",
}
# Or a list of rows
@pytest.fixture
def example_list_of_rows():
return [
{
"example_column_1": "entity_1",
"example_column_2": "entity_6",
"example_column_3": "biolink:related_to",
},
{
"example_column_1": "entity_2",
"example_column_2": "entity_7",
"example_column_3": "biolink:related_to",
},
]
# Define the mock koza transform
@pytest.fixture
def mock_transform(mock_koza, example_row):
return mock_koza(
INGEST_NAME,
example_row,
TRANSFORM_SCRIPT,
)
# Or for multiple rows
@pytest.fixture
def mock_transform_multiple_rows(mock_koza, example_list_of_rows):
return mock_koza(
INGEST_NAME,
example_list_of_rows,
TRANSFORM_SCRIPT,
)
# Test the output of the transform
def test_single_row(mock_transform):
assert len(mock_transform) == 1
entity = mock_transform[0]
assert entity
assert entity.subject == "entity_1"
def test_multiple_rows(mock_transform_multiple_rows):
assert len(mock_transform_multiple_rows) == 2
entity_1 = mock_transform_multiple_rows[0]
entity_2 = mock_transform_multiple_rows[1]
assert entity_1.subject == "entity_1"
assert entity_2.subject == "entity_2"