tests.system.openlineage.operator¶
Attributes¶
Classes¶
Which execution of a job an event belongs to. |
|
This operator is added for system testing purposes. |
Functions¶
|
|
|
Check values a task computes itself, given as |
|
|
|
|
|
|
|
|
|
|
|
Use this jinja method to access the environment variable named 'var'. |
|
|
|
|
|
|
|
Check if result is "equal" to expected value. |
|
Check that |
|
Order the events of executions by Dag run clear, loop run and iteration, map index and try, as |
Module Contents¶
- tests.system.openlineage.operator.check_values(checks)[source]¶
Check values a task computes itself, given as
{name: (actual, expected)}, e.g. what a macro returns.Every mismatch is logged as a validation error before failing the task.
- tests.system.openlineage.operator.env_var(var, default=None)[source]¶
Use this jinja method to access the environment variable named ‘var’.
If there is no such environment variable set, return the default. If the default is None, raise an exception for an undefined variable.
- tests.system.openlineage.operator.match(expected, result, env, path=None)[source]¶
Check if result is “equal” to expected value.
Omits keys not specified in expected value and resolves any jinja templates found.
Special sentinel values in expected dicts:
None(JSONnull) means the key must not be present in the result.{"$optional": "<reason>", ...}means the key is optional. When the key is present, the remaining template keys are validated. When the key is absent, validation passes silently and the reason is logged.{"$optional": "<reason>", "$value": <list_template>}is the list-template form of the optional sentinel. It behaves the same way, but also skips validation when the actual value is an empty list. Use this form when the template value is a list rather than a dict. Dict templates embed their keys directly alongside"$optional".
- class tests.system.openlineage.operator.Attempt[source]¶
Bases:
NamedTupleWhich execution of a job an event belongs to.
try_number,map_index,loop_indexandregion_idare None for a Dag event and when the task has none.clear_numberplays for a Dag run the roletry_numberplays for a task instance, andregion_idtells apart two executions of one loop iteration after the loop was cleared, which can share every other field.
- tests.system.openlineage.operator.check_run_id_integrity(events_by_key, repeated_start_keys=(), corrected_terminal_keys=(), unstarted_terminal_keys=())[source]¶
Check that
run.runIdidentifies exactly one attempt of exactly one job, and return the problems found.Events are grouped by
runId. The attempt of an event is its Dag run id and clear number, plus for a task event(try_number, map_index, loop.index, region_id), all fromrun.facets.airflow(airflowDagRunfor Dag events). Query events, which are child runs of a task, have no attempt. For every group:all events share one
job.name;all events share one attempt, so another Dag run, clear, retry, map index or loop iteration can’t reuse a
runId;the lifecycle is complete, so START and COMPLETE/FAIL of an attempt can’t carry different
runIdvalues:task and query: exactly one START, or several when the key of the START events is in
repeated_start_keys(deferral re-emits START on every resume), and exactly one COMPLETE or FAIL;Dag run: one START and at most one COMPLETE or FAIL. The Dag run that contains the checking task is still running when the check happens, so its COMPLETE/FAIL is missing, while Dag runs triggered by it can already have finished;
one more COMPLETE or FAIL is allowed when one of them has its key in
corrected_terminal_keys: a user marking a finished task or Dag run with another state emits a second one for the same run;no START is needed when the key of its COMPLETE or FAIL is in
unstarted_terminal_keys: a task that fails before it runs, e.g. on rendering its templates, or that a user marks before it started.
Across groups:
one attempt of a job has one
runId.
For a task in a loop, i.e. with
task.loop, also:runIdistaskInstance.id, as its iterations share every other coordinate;taskInstance.loop.region_idistaskInstance.region_idfor a task that isn’t mapped, and another region for a mapped one, whose own region is its expansion inside the loop’s;one run of a loop, i.e. its Dag run, clear and
loop.trajectory, has oneloop.region_id.
Only the events passed in are checked, i.e. those of the keys listed in the expected events.
- tests.system.openlineage.operator.sort_by_execution(event)[source]¶
Order the events of executions by Dag run clear, loop run and iteration, map index and try, as
event_sort_fn.
- class tests.system.openlineage.operator.OpenLineageTestOperator(event_templates=None, file_path=None, env=None, event_count_assertions=None, clear_variables=True, fail_fast=False, event_sort_fn=None, check_run_ids=True, corrected_terminal_keys=(), unstarted_terminal_keys=(), **kwargs)[source]¶
Bases:
airflow.providers.common.compat.sdk.BaseOperatorThis operator is added for system testing purposes.
It compares expected event templates set on initialization with ones emitted by OpenLineage integration and stored in Variables by VariableTransport.
- Note:
If clear_variables is True, only the Airflow Variables listed in event_templates (or derived from file_path) will be deleted - those that are supposed to be checked by the Operator. We won’t remove all Airflow Variables to avoid interfering with other instances of this Operator running in parallel on different DAGs. Running continuous system tests without clearing Variables may lead to leftover or growing Variables size. We recommend implementing a process to remove all Airflow Variables after all system tests have run to ensure a clean environment for each test run.
- Parameters:
event_templates (dict[str, dict | list] | None) – Dictionary where each key is the key used by VariableTransport, in the format
<DAG_ID>.<TASK_ID>.event.<EVENT_TYPE>. Each value may be one of the following: a single event template dict, matched against the last received event; a list of event template dicts, matched pairwise against all received events after sorting withevent_sort_fnwhen provided; or{"$absent": "<reason>"}, which asserts that no events exist for that key. The"$absent"reason is logged and included in the error if events are unexpectedly found.file_path (str | None) – alternatively, a path to a JSON file containing an array of event templates. Multiple events sharing the same
job.name+eventTypekey are accumulated into a list. An event with"$absent": "<reason>"asserts the variable must not exist. An event with"$since": "<version>"is checked only from that Airflow version on, e.g. when older versions don’t emit it.env (jinja2.Environment | None) – jinja environment used to render event templates.
event_count_assertions (dict[str, str | int] | None) – dict mapping a regex pattern (matched against variable keys) to a count expression. Supported expressions:
"N","==N",">=N","<=N". Applied to single-template keys only; list-template keys derive their count from the list length. Example:{".*\.event\.start": ">=2"}.clear_variables (bool) – if True, deletes only the checked variables after all checks run (or on failure). Default True.
fail_fast (bool) – if True, raise on the first failure. If False (default), collect all failures and raise a combined error at the end.
event_sort_fn (collections.abc.Callable[[dict], Any] | None) – optional callable
(event: dict) -> Anyused to sort actual events ascending before comparison. Useful for mapped tasks whose events arrive in non-deterministic order. When the template for a key is a list, sorting is required for pairwise comparison to be meaningful.corrected_terminal_keys (collections.abc.Collection[str]) – keys of COMPLETE/FAIL events emitted when a user marks a finished task or Dag run with another state. The run they belong to may then have one COMPLETE/FAIL more in the
check_run_id_integritycheck: the one of its original outcome, and the one of the correction.unstarted_terminal_keys (collections.abc.Collection[str]) – keys of COMPLETE/FAIL events of runs that have no START, as the task never ran: it failed before it started or a user marked it before it started. The
check_run_id_integritycheck then doesn’t require a START for their runs.check_run_ids (bool) – if True (default), additionally check the collected events with
check_run_id_integrity: START and COMPLETE/FAIL of one attempt share arunId, while another try, map index or loop iteration gets its own. Set to False only for a Dag whose expected events intentionally list a task’s lifecycle partially. It runs only once every key matches its template.
Special sentinel values in expected dicts:
None(JSONnull) means the key must not be present in the result.{"$optional": "<reason>", ...}means the key is optional. When the key is present, the remaining template keys are validated. When the key is absent, validation passes silently and the reason is logged.{"$optional": "<reason>", "$value": <list_template>}is the list-template form of the optional sentinel. It behaves the same way, but also skips validation when the actual value is an empty list. Use this form when the template value is a list rather than a dict. Dict templates embed their keys directly alongside"$optional".- Raises:
ValueError if the received events do not match with expected ones.