Hiiii, is there any example to write test for hamilton pipeline? Appreciate your help 🙏 😁

Hi @Duy Minh Thong Nguyen! I don't think we have a documentation page / example on the topic (yet). With Hamilton, there is generally two levels of testing: 1. unit test 2. "dataflow / DAG" test ## 1. Unit test Since Hamilton, uses regular Python functions, you can use pytest to test individual nodes in your graph. It should check that the code ensures "output Y for input X". Typically, you don't need to import Hamilton in these tests. Here we check that

features()

returns a dataframe with column

user_id

# feature_processing.py
import pandas as pd

def raw_data(file_path: str) -> pd.DataFrame:
  return pd.read_csv(file_path)

def features(raw_data: pd.DataFrame) -> pd.DataFrame:
  # do some transforms
  return features

# test_feature_processing_nodes.py
import feature_processing

def test_features_include_user_id():
  test_input = pd.DataFrame(...)
  result = feature_processing.features(raw_data=test_input)
  assert "user_id" in result.columns

## 2. dataflow test Here the goal is to check that changes to the dataflow and nodes didn't break important paths or properties. Here, you will import Hamilton. For example, you could check that the node

trained_model

exist and that the path between `file_path`and

trained_model

is valid. You could also test things around tags, etc.

# test_feature_processing_dataflow.py
from hamilton import driver
import feature_processing

def test_trained_model_exists():
  dr = driver.Builder().with_modules(feature_processing).build()
  assert "trained_model" in dr.list_available_nodes()

def test_path_raw_data_to_trained_model():
  dr = driver.Builder().with_modules(feature_processing).build()
  assert dr.validate_execution(["trained_model"], inputs={"file_path": ...})

I hope this gives some context. With more context on your Hamilton use case, I can give you more pointers!

Hi @Thierry Jean, I want to create test to guarantee the correctness of pipeline. For example, I want to check this:

def A(items: list[dict]) -> Parallelizable[dict]:
    for item in items:
        yield item

def B(A: dict) -> str:
    return A["key"]

def C(A: dict) -> str:
    return A["value"]

def D(B: str, C: str) -> str:
    # Is B and C guaranteed to be the same item?
    pass

If I understand, your test should look something like this:

import my_pipeline

def test_B_and_C_are_equal():
   input_A = {"key": 0, "value": 1}
   assert my_pipeline.B(A=input_A) == my_pipeline.C(A=input_A)

However, in your code,

B()

has

A["key"]

while

C()

has

A["value"]

. Do you expect them to have the same or different results?

I expect them to belong to one object

A = [{"key": 0, "value": 0}, {"key": 1, "value": 1}, {"key": 2, "value": 2}] (all pairs are identical)

I'm concerned that the merge process in parallel might cause mismatches at D such as (2, 1) or (0, 1) where keys and values don't match up as they should 👀

So the way

Parallelizable/Collect

work, the

item

object yielded by

A

(function annotated with

Parallelizable

) will be the same object across nodes until the

Collect

function. In other words, you should never have mismatch `(2, 1)`; Hamilton will always guarantee

(0, 0)

,

(1, 1)

, etc. What Hamilton doesn't guarantee currently is that items will be processed in a deterministic order. Maybe the first time you run your code, you will run

(0, 0), (1,1), (3, 3), (4, 4), (2, 2)

but if you rerun your code, you will get

(0,0), (2, 2), (1, 1), (3, 3), (4, 4)

. Does that help?