mirror of
https://github.com/datahub-project/datahub.git
synced 2025-07-24 10:00:07 +00:00
24 lines
812 B
Python
24 lines
812 B
Python
|
import datahub.emitter.mce_builder as builder
|
||
|
|
from datahub.metadata.schema_classes import (
|
||
|
|
DataFlowInfoClass,
|
||
|
|
DatasetPropertiesClass,
|
||
|
|
DatasetSnapshotClass,
|
||
|
|
MetadataChangeEventClass,
|
||
|
|
OwnershipClass,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_can_add_aspect():
|
||
|
|
dataset_mce: MetadataChangeEventClass = builder.make_lineage_mce(
|
||
|
|
[
|
||
|
|
builder.make_dataset_urn("bigquery", "upstream1"),
|
||
|
|
builder.make_dataset_urn("bigquery", "upstream2"),
|
||
|
|
],
|
||
|
|
builder.make_dataset_urn("bigquery", "downstream"),
|
||
|
|
)
|
||
|
|
assert isinstance(dataset_mce.proposedSnapshot, DatasetSnapshotClass)
|
||
|
|
|
||
|
|
assert builder.can_add_aspect(dataset_mce, DatasetPropertiesClass)
|
||
|
|
assert builder.can_add_aspect(dataset_mce, OwnershipClass)
|
||
|
|
assert not builder.can_add_aspect(dataset_mce, DataFlowInfoClass)
|