diff --git a/fixtures/golden/adrema-audience-selection/expected-output.json b/fixtures/golden/adrema-audience-selection/expected-output.json new file mode 100644 index 0000000..496cc0e --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/expected-output.json @@ -0,0 +1,70 @@ +[ + { + "recipient_key": "identity:alice", + "display_name": "Alice Example", + "identity_id": "identity-alice", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": "alice@example.test", + "postal_address": null, + "account_id": null, + "portal_target": null, + "contact_point_id": "contact-email-alice", + "selected_channel": "email", + "distribution_status": "usable", + "exclusion_reason": null, + "policy_decision": "preference.explicit_email", + "_reconciliation_status": "match" + }, + { + "recipient_key": "identity:bob", + "display_name": "Bob Example", + "identity_id": "identity-bob", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": null, + "postal_address": "Example Street 2, 10000 Exampletown", + "account_id": null, + "portal_target": null, + "contact_point_id": "contact-postal-bob", + "selected_channel": "postal", + "distribution_status": "suppressed", + "exclusion_reason": "preference.suppressed", + "policy_decision": "preference.suppressed", + "_reconciliation_status": "match" + }, + { + "recipient_key": "identity:grace", + "display_name": "Grace Example", + "identity_id": "identity-grace", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": null, + "postal_address": "Example Street 3, 10000 Exampletown", + "account_id": null, + "portal_target": null, + "contact_point_id": "contact-postal-grace", + "selected_channel": "postal", + "distribution_status": "usable", + "exclusion_reason": null, + "policy_decision": "preference.default_postal", + "_reconciliation_status": "match" + }, + { + "recipient_key": "identity:frank", + "display_name": "Frank Example", + "identity_id": "identity-frank", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": null, + "postal_address": null, + "account_id": null, + "portal_target": "portal:frank", + "contact_point_id": "contact-portal-frank", + "selected_channel": "portal", + "distribution_status": "usable", + "exclusion_reason": null, + "policy_decision": "manual.include_with_reason", + "_reconciliation_status": "match" + } +] diff --git a/fixtures/golden/adrema-audience-selection/graph.json b/fixtures/golden/adrema-audience-selection/graph.json new file mode 100644 index 0000000..b4d0717 --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/graph.json @@ -0,0 +1,299 @@ +{ + "schema_version": 1, + "nodes": [ + { + "id": "directory", + "type": "source.reference", + "label": "Frozen directory and contact snapshot", + "position": {"x": 40, "y": 60}, + "config": { + "source_ref": "snapshot:directory-2026-01", + "source_name": "directory_snapshot", + "expected_fingerprint": "sha256:synthetic-directory-2026-01", + "consistency": "frozen", + "source_columns": [ + "recipient_key", "display_name", "identity_id", "group_key", + "organization_unit_id", "function_id", "region", "purpose", + "effective", "effective_from", "effective_until", + "preferred_target", "preferred_contact_point_id", "preferred_channel", + "fallback_target", "fallback_contact_point_id", "fallback_channel", + "preference_decision", "sort_order" + ] + } + }, + { + "id": "group-filter", + "type": "filter", + "label": "Typed group", + "position": {"x": 260, "y": 60}, + "config": {"column": "group_key", "operator": "eq", "value": {"$parameter": "group_key"}} + }, + { + "id": "organization-filter", + "type": "filter", + "label": "Organization unit", + "position": {"x": 480, "y": 60}, + "config": {"column": "organization_unit_id", "operator": "eq", "value": "district-north"} + }, + { + "id": "function-filter", + "type": "filter", + "label": "Effective function", + "position": {"x": 700, "y": 60}, + "config": {"column": "function_id", "operator": "eq", "value": "resident-service"} + }, + { + "id": "region-filter", + "type": "filter", + "label": "Geography", + "position": {"x": 920, "y": 60}, + "config": {"column": "region", "operator": "eq", "value": {"$parameter": "region"}} + }, + { + "id": "purpose-filter", + "type": "filter", + "label": "Distribution purpose", + "position": {"x": 1140, "y": 60}, + "config": {"column": "purpose", "operator": "eq", "value": {"$parameter": "purpose"}} + }, + { + "id": "effective-filter", + "type": "filter", + "label": "Effective-dated assignment", + "position": {"x": 1360, "y": 60}, + "config": {"column": "effective", "operator": "eq", "value": true} + }, + { + "id": "contact-fallback", + "type": "calculate", + "label": "Resolve preferred or fallback contact", + "position": {"x": 1580, "y": 60}, + "config": { + "calculations": [ + {"target_column": "selected_target", "expression": "coalesce(preferred_target, fallback_target)", "result_type": "string"}, + {"target_column": "selected_contact_point_id", "expression": "coalesce(preferred_contact_point_id, fallback_contact_point_id)", "result_type": "string"}, + {"target_column": "selected_channel", "expression": "coalesce(preferred_channel, fallback_channel)", "result_type": "string"}, + {"target_column": "email", "expression": "CASE WHEN selected_channel = 'email' THEN selected_target ELSE NULL END", "result_type": "string"}, + {"target_column": "postal_address", "expression": "CASE WHEN selected_channel = 'postal' THEN selected_target ELSE NULL END", "result_type": "string"}, + {"target_column": "account_id", "expression": "CASE WHEN selected_channel = 'internal_mail' THEN selected_target ELSE NULL END", "result_type": "string"}, + {"target_column": "portal_target", "expression": "CASE WHEN selected_channel = 'portal' THEN selected_target ELSE NULL END", "result_type": "string"} + ] + } + }, + { + "id": "directory-fields", + "type": "select", + "label": "Canonical recipient candidate", + "position": {"x": 1800, "y": 60}, + "config": { + "fields": [ + "recipient_key", "display_name", "identity_id", "group_key", + "organization_unit_id", "function_id", "selected_target", + "selected_contact_point_id", "selected_channel", "email", + "postal_address", "account_id", "portal_target", + "preference_decision", "sort_order" + ] + } + }, + { + "id": "candidate-quality", + "type": "quality.rules", + "label": "Recipient quality gate", + "position": {"x": 2020, "y": 60}, + "config": { + "rules": [ + {"id": "recipient-key", "column": "recipient_key", "operator": "not_null"}, + {"id": "delivery-target", "column": "selected_target", "operator": "not_null"}, + {"id": "contact-point", "column": "selected_contact_point_id", "operator": "not_null"} + ], + "action": "drop" + } + }, + { + "id": "manual", + "type": "source.reference", + "label": "Frozen manual includes", + "position": {"x": 1580, "y": 300}, + "config": { + "source_ref": "snapshot:manual-includes-2026-01", + "source_name": "manual_includes", + "expected_fingerprint": "sha256:synthetic-manual-includes-2026-01", + "consistency": "frozen", + "source_columns": [ + "recipient_key", "display_name", "identity_id", "group_key", + "organization_unit_id", "function_id", "selected_target", + "selected_contact_point_id", "selected_channel", "email", + "postal_address", "account_id", "portal_target", + "preference_decision", "sort_order" + ] + } + }, + { + "id": "append-includes", + "type": "combine.union", + "label": "Append manual includes", + "position": {"x": 2240, "y": 170}, + "config": {"mode": "all"} + }, + { + "id": "deduplicate", + "type": "distinct", + "label": "One candidate per identity", + "position": {"x": 2460, "y": 170}, + "config": {"columns": ["recipient_key"]} + }, + { + "id": "suppressions", + "type": "source.reference", + "label": "Frozen channel suppressions", + "position": {"x": 2240, "y": 410}, + "config": { + "source_ref": "snapshot:suppressions-2026-01", + "source_name": "channel_suppressions", + "expected_fingerprint": "sha256:synthetic-suppressions-2026-01", + "consistency": "frozen", + "source_columns": ["recipient_key", "suppression_reason"] + } + }, + { + "id": "apply-suppressions", + "type": "combine.join", + "label": "Attach suppression decisions", + "position": {"x": 2680, "y": 250}, + "config": { + "join_type": "left", + "left_keys": ["recipient_key"], + "right_keys": ["recipient_key"], + "right_prefix": "suppression_" + } + }, + { + "id": "decision-columns", + "type": "calculate", + "label": "Classify channel decision", + "position": {"x": 2900, "y": 250}, + "config": { + "calculations": [ + {"target_column": "distribution_status", "expression": "CASE WHEN suppression_suppression_reason IS NOT NULL THEN 'suppressed' ELSE 'usable' END", "result_type": "string"}, + {"target_column": "exclusion_reason", "expression": "suppression_suppression_reason", "result_type": "string"}, + {"target_column": "policy_decision", "expression": "coalesce(suppression_suppression_reason, preference_decision)", "result_type": "string"} + ] + } + }, + { + "id": "decision-fields", + "type": "select", + "label": "Canonical audience decision", + "position": {"x": 3120, "y": 250}, + "config": { + "fields": [ + "recipient_key", "display_name", "identity_id", + "organization_unit_id", "function_id", "email", "postal_address", + "account_id", "portal_target", + {"column": "selected_contact_point_id", "alias": "contact_point_id"}, + "selected_channel", "distribution_status", "exclusion_reason", + "policy_decision", "sort_order" + ] + } + }, + { + "id": "stable-order", + "type": "sort", + "label": "Stable recipient order", + "position": {"x": 3340, "y": 250}, + "config": { + "fields": [ + {"column": "sort_order", "direction": "asc"}, + {"column": "recipient_key", "direction": "asc"} + ] + } + }, + { + "id": "expected", + "type": "source.reference", + "label": "Frozen expected audience", + "position": {"x": 3340, "y": 500}, + "config": { + "source_ref": "snapshot:expected-audience-2026-01", + "source_name": "expected_audience", + "expected_fingerprint": "sha256:synthetic-expected-audience-2026-01", + "consistency": "frozen", + "source_columns": [ + "recipient_key", "display_name", "identity_id", + "organization_unit_id", "function_id", "email", "postal_address", + "account_id", "portal_target", "contact_point_id", + "selected_channel", "distribution_status", "exclusion_reason", + "policy_decision" + ] + } + }, + { + "id": "reconcile", + "type": "reconcile.compare", + "label": "Reconcile expected audience", + "position": {"x": 3560, "y": 350}, + "config": { + "left_keys": ["recipient_key"], + "right_keys": ["recipient_key"], + "compare_columns": [ + "selected_channel", "contact_point_id", "distribution_status", + "exclusion_reason", "policy_decision" + ], + "right_prefix": "actual_" + } + }, + { + "id": "reconciled-only", + "type": "filter", + "label": "Require exact reconciliation", + "position": {"x": 3780, "y": 350}, + "config": {"column": "_reconciliation_status", "operator": "eq", "value": "match"} + }, + { + "id": "output-fields", + "type": "select", + "label": "Distribution-list dataset", + "position": {"x": 4000, "y": 350}, + "config": { + "fields": [ + "recipient_key", "display_name", "identity_id", + "organization_unit_id", "function_id", "email", "postal_address", + "account_id", "portal_target", "contact_point_id", + "selected_channel", "distribution_status", "exclusion_reason", + "policy_decision", "_reconciliation_status" + ] + } + }, + { + "id": "output", + "type": "output", + "label": "Pinned audience output", + "position": {"x": 4220, "y": 350}, + "config": {} + } + ], + "edges": [ + {"id": "e01", "source": "directory", "target": "group-filter"}, + {"id": "e02", "source": "group-filter", "target": "organization-filter"}, + {"id": "e03", "source": "organization-filter", "target": "function-filter"}, + {"id": "e04", "source": "function-filter", "target": "region-filter"}, + {"id": "e05", "source": "region-filter", "target": "purpose-filter"}, + {"id": "e06", "source": "purpose-filter", "target": "effective-filter"}, + {"id": "e07", "source": "effective-filter", "target": "contact-fallback"}, + {"id": "e08", "source": "contact-fallback", "target": "directory-fields"}, + {"id": "e09", "source": "directory-fields", "target": "candidate-quality"}, + {"id": "e10", "source": "candidate-quality", "target": "append-includes"}, + {"id": "e11", "source": "manual", "target": "append-includes"}, + {"id": "e12", "source": "append-includes", "target": "deduplicate"}, + {"id": "e13", "source": "deduplicate", "target": "apply-suppressions", "target_port": "left"}, + {"id": "e14", "source": "suppressions", "target": "apply-suppressions", "target_port": "right"}, + {"id": "e15", "source": "apply-suppressions", "target": "decision-columns"}, + {"id": "e16", "source": "decision-columns", "target": "decision-fields"}, + {"id": "e17", "source": "decision-fields", "target": "stable-order"}, + {"id": "e18", "source": "expected", "target": "reconcile", "target_port": "left"}, + {"id": "e19", "source": "stable-order", "target": "reconcile", "target_port": "right"}, + {"id": "e20", "source": "reconcile", "target": "reconciled-only"}, + {"id": "e21", "source": "reconciled-only", "target": "output-fields"}, + {"id": "e22", "source": "output-fields", "target": "output"} + ] +} diff --git a/fixtures/golden/adrema-audience-selection/inputs/directory-snapshot.json b/fixtures/golden/adrema-audience-selection/inputs/directory-snapshot.json new file mode 100644 index 0000000..6d082bf --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/inputs/directory-snapshot.json @@ -0,0 +1,149 @@ +[ + { + "recipient_key": "identity:alice", + "display_name": "Alice Example", + "identity_id": "identity-alice", + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "region": "north", + "purpose": "monthly-notice", + "effective": true, + "effective_from": "2025-01-01", + "effective_until": null, + "preferred_target": "alice@example.test", + "preferred_contact_point_id": "contact-email-alice", + "preferred_channel": "email", + "fallback_target": "Example Street 1, 10000 Exampletown", + "fallback_contact_point_id": "contact-postal-alice", + "fallback_channel": "postal", + "preference_decision": "preference.explicit_email", + "sort_order": 10 + }, + { + "recipient_key": "identity:bob", + "display_name": "Bob Example", + "identity_id": "identity-bob", + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "region": "north", + "purpose": "monthly-notice", + "effective": true, + "effective_from": "2025-06-01", + "effective_until": "2027-01-01", + "preferred_target": null, + "preferred_contact_point_id": null, + "preferred_channel": null, + "fallback_target": "Example Street 2, 10000 Exampletown", + "fallback_contact_point_id": "contact-postal-bob", + "fallback_channel": "postal", + "preference_decision": "preference.default_postal", + "sort_order": 20 + }, + { + "recipient_key": "identity:grace", + "display_name": "Grace Example", + "identity_id": "identity-grace", + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "region": "north", + "purpose": "monthly-notice", + "effective": true, + "effective_from": "2024-01-01", + "effective_until": null, + "preferred_target": null, + "preferred_contact_point_id": null, + "preferred_channel": null, + "fallback_target": "Example Street 3, 10000 Exampletown", + "fallback_contact_point_id": "contact-postal-grace", + "fallback_channel": "postal", + "preference_decision": "preference.default_postal", + "sort_order": 30 + }, + { + "recipient_key": "identity:carol", + "display_name": "Carol Example", + "identity_id": "identity-carol", + "group_key": "employees", + "organization_unit_id": "district-south", + "function_id": "finance-officer", + "region": "south", + "purpose": "monthly-notice", + "effective": true, + "effective_from": "2025-01-01", + "effective_until": null, + "preferred_target": "carol@example.test", + "preferred_contact_point_id": "contact-email-carol", + "preferred_channel": "email", + "fallback_target": null, + "fallback_contact_point_id": null, + "fallback_channel": null, + "preference_decision": "preference.explicit_email", + "sort_order": 40 + }, + { + "recipient_key": "identity:dana", + "display_name": "Dana Example", + "identity_id": "identity-dana", + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "region": "north", + "purpose": "newsletter", + "effective": true, + "effective_from": "2025-01-01", + "effective_until": null, + "preferred_target": "dana@example.test", + "preferred_contact_point_id": "contact-email-dana", + "preferred_channel": "email", + "fallback_target": null, + "fallback_contact_point_id": null, + "fallback_channel": null, + "preference_decision": "preference.explicit_email", + "sort_order": 50 + }, + { + "recipient_key": "identity:erin", + "display_name": "Erin Example", + "identity_id": "identity-erin", + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "region": "north", + "purpose": "monthly-notice", + "effective": false, + "effective_from": "2024-01-01", + "effective_until": "2025-01-01", + "preferred_target": "erin@example.test", + "preferred_contact_point_id": "contact-email-erin", + "preferred_channel": "email", + "fallback_target": null, + "fallback_contact_point_id": null, + "fallback_channel": null, + "preference_decision": "preference.explicit_email", + "sort_order": 60 + }, + { + "recipient_key": null, + "display_name": "Malformed Synthetic Row", + "identity_id": null, + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "region": "north", + "purpose": "monthly-notice", + "effective": true, + "effective_from": "2025-01-01", + "effective_until": null, + "preferred_target": null, + "preferred_contact_point_id": null, + "preferred_channel": null, + "fallback_target": null, + "fallback_contact_point_id": null, + "fallback_channel": null, + "preference_decision": "preference.unresolved", + "sort_order": 70 + } +] diff --git a/fixtures/golden/adrema-audience-selection/inputs/expected-audience.json b/fixtures/golden/adrema-audience-selection/inputs/expected-audience.json new file mode 100644 index 0000000..6755db5 --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/inputs/expected-audience.json @@ -0,0 +1,66 @@ +[ + { + "recipient_key": "identity:alice", + "display_name": "Alice Example", + "identity_id": "identity-alice", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": "alice@example.test", + "postal_address": null, + "account_id": null, + "portal_target": null, + "contact_point_id": "contact-email-alice", + "selected_channel": "email", + "distribution_status": "usable", + "exclusion_reason": null, + "policy_decision": "preference.explicit_email" + }, + { + "recipient_key": "identity:bob", + "display_name": "Bob Example", + "identity_id": "identity-bob", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": null, + "postal_address": "Example Street 2, 10000 Exampletown", + "account_id": null, + "portal_target": null, + "contact_point_id": "contact-postal-bob", + "selected_channel": "postal", + "distribution_status": "suppressed", + "exclusion_reason": "preference.suppressed", + "policy_decision": "preference.suppressed" + }, + { + "recipient_key": "identity:grace", + "display_name": "Grace Example", + "identity_id": "identity-grace", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": null, + "postal_address": "Example Street 3, 10000 Exampletown", + "account_id": null, + "portal_target": null, + "contact_point_id": "contact-postal-grace", + "selected_channel": "postal", + "distribution_status": "usable", + "exclusion_reason": null, + "policy_decision": "preference.default_postal" + }, + { + "recipient_key": "identity:frank", + "display_name": "Frank Example", + "identity_id": "identity-frank", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "email": null, + "postal_address": null, + "account_id": null, + "portal_target": "portal:frank", + "contact_point_id": "contact-portal-frank", + "selected_channel": "portal", + "distribution_status": "usable", + "exclusion_reason": null, + "policy_decision": "manual.include_with_reason" + } +] diff --git a/fixtures/golden/adrema-audience-selection/inputs/manual-includes.json b/fixtures/golden/adrema-audience-selection/inputs/manual-includes.json new file mode 100644 index 0000000..827e6d1 --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/inputs/manual-includes.json @@ -0,0 +1,36 @@ +[ + { + "recipient_key": "identity:alice", + "display_name": "Alice Example (duplicate manual include)", + "identity_id": "identity-alice", + "group_key": "residents", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "selected_target": "alice@example.test", + "selected_contact_point_id": "contact-email-alice", + "selected_channel": "email", + "email": "alice@example.test", + "postal_address": null, + "account_id": null, + "portal_target": null, + "preference_decision": "manual.include_duplicate", + "sort_order": 11 + }, + { + "recipient_key": "identity:frank", + "display_name": "Frank Example", + "identity_id": "identity-frank", + "group_key": "manual", + "organization_unit_id": "district-north", + "function_id": "resident-service", + "selected_target": "portal:frank", + "selected_contact_point_id": "contact-portal-frank", + "selected_channel": "portal", + "email": null, + "postal_address": null, + "account_id": null, + "portal_target": "portal:frank", + "preference_decision": "manual.include_with_reason", + "sort_order": 40 + } +] diff --git a/fixtures/golden/adrema-audience-selection/inputs/suppressions.json b/fixtures/golden/adrema-audience-selection/inputs/suppressions.json new file mode 100644 index 0000000..2672dc8 --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/inputs/suppressions.json @@ -0,0 +1,6 @@ +[ + { + "recipient_key": "identity:bob", + "suppression_reason": "preference.suppressed" + } +] diff --git a/fixtures/golden/adrema-audience-selection/manifest.json b/fixtures/golden/adrema-audience-selection/manifest.json new file mode 100644 index 0000000..1ff6b7a --- /dev/null +++ b/fixtures/golden/adrema-audience-selection/manifest.json @@ -0,0 +1,92 @@ +{ + "id": "adrema-audience-selection", + "title": "Synthetic governed AdreMa audience selection", + "issue": "GovOPlaN/govoplan-dataflow#18", + "tenant_id": "tenant-synthetic-adrema", + "graph": "graph.json", + "expected_output": "expected-output.json", + "parameters": { + "group_key": "residents", + "region": "north", + "purpose": "monthly-notice" + }, + "sources": { + "snapshot:directory-2026-01": { + "fixture": "inputs/directory-snapshot.json", + "tenant_id": "tenant-synthetic-adrema", + "provider": "connectors.synthetic-directory", + "fingerprint": "sha256:synthetic-directory-2026-01" + }, + "snapshot:manual-includes-2026-01": { + "fixture": "inputs/manual-includes.json", + "tenant_id": "tenant-synthetic-adrema", + "provider": "datasources.static", + "fingerprint": "sha256:synthetic-manual-includes-2026-01" + }, + "snapshot:suppressions-2026-01": { + "fixture": "inputs/suppressions.json", + "tenant_id": "tenant-synthetic-adrema", + "provider": "addresses.synthetic-preferences", + "fingerprint": "sha256:synthetic-suppressions-2026-01" + }, + "snapshot:expected-audience-2026-01": { + "fixture": "inputs/expected-audience.json", + "tenant_id": "tenant-synthetic-adrema", + "provider": "reporting.synthetic-baseline", + "fingerprint": "sha256:synthetic-expected-audience-2026-01" + } + }, + "expected_definition_hash": "2fec104dfda522dbc708ce61b237c062919703d62d2912ba621d5873ee5b975e", + "expected_execution_graph_hash": "d1cbb51c837810300044ed5f660e295ace49411265108b102731174d5587a85d", + "expected_output_hash": "f2d744354279f80393c6f894f7854569aa686f8292e6256fe429569d4410a72e", + "expected_source_fingerprints": [ + { + "node_id": "directory", + "source_ref": "snapshot:directory-2026-01", + "source_name": "directory_snapshot", + "kind": "datasource", + "provider": "connectors.synthetic-directory", + "fingerprint": "sha256:synthetic-directory-2026-01", + "row_count": 7, + "preview_rows": 7, + "truncated": false + }, + { + "node_id": "manual", + "source_ref": "snapshot:manual-includes-2026-01", + "source_name": "manual_includes", + "kind": "datasource", + "provider": "datasources.static", + "fingerprint": "sha256:synthetic-manual-includes-2026-01", + "row_count": 2, + "preview_rows": 2, + "truncated": false + }, + { + "node_id": "suppressions", + "source_ref": "snapshot:suppressions-2026-01", + "source_name": "channel_suppressions", + "kind": "datasource", + "provider": "addresses.synthetic-preferences", + "fingerprint": "sha256:synthetic-suppressions-2026-01", + "row_count": 1, + "preview_rows": 1, + "truncated": false + }, + { + "node_id": "expected", + "source_ref": "snapshot:expected-audience-2026-01", + "source_name": "expected_audience", + "kind": "datasource", + "provider": "reporting.synthetic-baseline", + "fingerprint": "sha256:synthetic-expected-audience-2026-01", + "row_count": 4, + "preview_rows": 4, + "truncated": false + } + ], + "expected_recipients": ["identity:alice", "identity:grace", "identity:frank"], + "expected_exclusions": { + "identity:bob": "preference.suppressed" + } +} diff --git a/src/govoplan_dataflow/backend/dataset_output.py b/src/govoplan_dataflow/backend/dataset_output.py new file mode 100644 index 0000000..d68e26b --- /dev/null +++ b/src/govoplan_dataflow/backend/dataset_output.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import hashlib +import json + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.dataflows import ( + DataflowDatasetDescriptor, + DataflowDatasetRequest, + DataflowDatasetResult, + DataflowRunConflictError, + DataflowRunUnavailableError, +) +from govoplan_core.security.time import utc_now +from govoplan_dataflow.backend.backends.base import ExecutionBudget +from govoplan_dataflow.backend.executor import PipelineExecutionError +from govoplan_dataflow.backend.governance import require_definition_action +from govoplan_dataflow.backend.schemas import PipelineGraph +from govoplan_dataflow.backend.service import ( + _execute_pipeline_preview, + get_pipeline, + get_pipeline_revision, + list_pipelines, +) +from govoplan_dataflow.backend.subflows import substitute_parameters + + +class SqlDataflowDatasetOutputProvider: + def __init__(self, registry: object | None = None) -> None: + self.registry = registry + + def list_outputs( + self, + session: object, + principal: object, + *, + query: str = "", + limit: int = 100, + ) -> tuple[DataflowDatasetDescriptor, ...]: + typed_session, typed_principal = _contracts(session, principal) + normalized_query = query.strip().casefold() + result: list[DataflowDatasetDescriptor] = [] + for pipeline in list_pipelines( + typed_session, + tenant_id=typed_principal.tenant_id, + ): + if normalized_query and normalized_query not in pipeline.name.casefold(): + continue + try: + decision = require_definition_action( + pipeline, + principal=typed_principal, + registry=self.registry, + action="run", + ) + except PermissionError: + continue + revision = get_pipeline_revision(typed_session, pipeline=pipeline) + result.append( + DataflowDatasetDescriptor( + pipeline_ref=pipeline.id, + name=pipeline.name, + description=pipeline.description, + revision=revision.revision, + definition_hash=revision.content_hash, + status=pipeline.status, + updated_at=pipeline.updated_at, + parameters=dict(pipeline.metadata_.get("parameters") or {}), + provenance={ + "module": "dataflow", + "scope_type": pipeline.scope_type, + "scope_id": pipeline.scope_id, + "policy_decision": decision.to_dict(), + }, + ) + ) + if len(result) >= max(1, min(limit, 200)): + break + return tuple(result) + + def read_output( + self, + session: object, + principal: object, + *, + request: DataflowDatasetRequest, + ) -> DataflowDatasetResult: + typed_session, typed_principal = _contracts(session, principal) + row_limit = max(1, min(request.row_limit, 2_000)) + pipeline = get_pipeline( + typed_session, + tenant_id=typed_principal.tenant_id, + pipeline_id=request.pipeline_ref, + ) + decision = require_definition_action( + pipeline, + principal=typed_principal, + registry=self.registry, + action="run", + ) + revision = get_pipeline_revision( + typed_session, + pipeline=pipeline, + revision=request.revision, + ) + if ( + request.expected_definition_hash + and request.expected_definition_hash != revision.content_hash + ): + raise DataflowRunConflictError( + "The pinned Dataflow definition hash no longer matches the requested revision." + ) + graph = PipelineGraph.model_validate( + substitute_parameters(revision.graph, dict(request.parameters)) + ) + try: + result, executor_version = _execute_pipeline_preview( + graph, + session=typed_session, + principal=typed_principal, + registry=self.registry, + backend="auto", + row_limit=row_limit, + preview_node_id=None, + budget=ExecutionBudget( + max_output_rows=row_limit, + max_batch_bytes=4_000_000, + max_wall_seconds=5.0, + ), + ) + except PipelineExecutionError as exc: + raise DataflowRunUnavailableError( + f"The pinned Dataflow output could not be evaluated: {exc}" + ) from exc + source_fingerprints = tuple( + dict(item) for item in result.source_fingerprints + ) + if request.expected_source_fingerprints and not _fingerprints_match( + request.expected_source_fingerprints, + source_fingerprints, + ): + raise DataflowRunConflictError( + "Dataflow source fingerprints differ from the pinned source state." + ) + output_hash = hashlib.sha256( + json.dumps( + result.rows, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + default=str, + ).encode("utf-8") + ).hexdigest() + return DataflowDatasetResult( + pipeline_ref=pipeline.id, + revision=revision.revision, + definition_hash=revision.content_hash, + rows=tuple(dict(item) for item in result.rows), + total_rows=result.total_rows, + truncated=result.truncated, + output_hash=output_hash, + executor_version=executor_version, + source_fingerprints=source_fingerprints, + diagnostics=tuple( + item.model_dump(mode="json") for item in result.diagnostics + ), + generated_at=utc_now(), + provenance={ + "module": "dataflow", + "scope_type": pipeline.scope_type, + "scope_id": pipeline.scope_id, + "policy_decision": decision.to_dict(), + "parameters": dict(request.parameters), + }, + ) + + +def dataset_output_provider(context: object | None = None): + return SqlDataflowDatasetOutputProvider(getattr(context, "registry", None)) + + +def _fingerprints_match(expected, actual) -> bool: + def normalized(values): + return sorted( + json.dumps(dict(item), sort_keys=True, separators=(",", ":"), default=str) + for item in values + ) + + return normalized(expected) == normalized(actual) + + +def _contracts(session: object, principal: object): + if not hasattr(session, "scalar") or not hasattr(session, "scalars"): + raise TypeError("Dataflow dataset output requires a SQLAlchemy session.") + if not isinstance(principal, ApiPrincipal): + raise TypeError("Dataflow dataset output requires an API principal.") + return session, principal + + +__all__ = ["SqlDataflowDatasetOutputProvider", "dataset_output_provider"] diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 32d1785..facb7d2 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -11,6 +11,7 @@ from govoplan_core.core.access import ( CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, ) from govoplan_core.core.dataflows import ( + CAPABILITY_DATAFLOW_DATASET_OUTPUT, CAPABILITY_DATAFLOW_RUN_LIFECYCLE, CAPABILITY_DATAFLOW_RUN_WORKER, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, @@ -279,7 +280,7 @@ manifest = ModuleManifest( ModuleInterfaceProvider(name="dataflow.pipeline_preview", version=MODULE_VERSION), ModuleInterfaceProvider(name="dataflow.run_lifecycle", version=MODULE_VERSION), ModuleInterfaceProvider(name="dataflow.run_worker", version=MODULE_VERSION), - ModuleInterfaceProvider(name="dataflow.dataset_output", version=MODULE_VERSION), + ModuleInterfaceProvider(name=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION), ModuleInterfaceProvider( name="dataflow.trigger_dispatcher", version=MODULE_VERSION, @@ -378,6 +379,10 @@ manifest = ModuleManifest( ), route_factory=_dataflow_router, capability_factories={ + CAPABILITY_DATAFLOW_DATASET_OUTPUT: lambda context: __import__( + "govoplan_dataflow.backend.dataset_output", + fromlist=["dataset_output_provider"], + ).dataset_output_provider(context), CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider, CAPABILITY_DATAFLOW_RUN_WORKER: _run_worker, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider, diff --git a/tests/test_dataset_output.py b/tests/test_dataset_output.py new file mode 100644 index 0000000..3e2f5e2 --- /dev/null +++ b/tests/test_dataset_output.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.dataflows import DataflowDatasetRequest, DataflowRunConflictError +from govoplan_core.db.base import Base +from govoplan_core.db.session import configure_database, reset_database +from govoplan_dataflow.backend.dataset_output import SqlDataflowDatasetOutputProvider +from govoplan_dataflow.backend.db.models import DataflowPipeline, DataflowPipelineRevision +from govoplan_dataflow.backend.schemas import PipelineGraph +from govoplan_dataflow.backend.service import definition_hash + + +def principal(tenant_id: str = "tenant-1") -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id=tenant_id, + scopes=frozenset({"dataflow:pipeline:read", "dataflow:pipeline:run"}), + ), + account=object(), + user=object(), + ) + + +class DataflowDatasetOutputTests(unittest.TestCase): + def setUp(self) -> None: + self.database = configure_database("sqlite:///:memory:") + Base.metadata.create_all( + self.database.engine, + tables=[DataflowPipeline.__table__, DataflowPipelineRevision.__table__], + ) + + def tearDown(self) -> None: + reset_database(dispose=True) + + def test_pinned_revision_executes_through_typed_registry(self) -> None: + graph = PipelineGraph.model_validate( + { + "nodes": [ + { + "id": "source", + "type": "source.inline", + "label": "Source", + "position": {"x": 0, "y": 0}, + "config": { + "source_name": "audience", + "rows": [ + {"recipient_key": "one", "email": "one@example.test"}, + {"recipient_key": "two", "email": "two@example.test"}, + ] + }, + }, + { + "id": "output", + "type": "output", + "label": "Output", + "position": {"x": 200, "y": 0}, + "config": {}, + }, + ], + "edges": [{"id": "edge-1", "source": "source", "target": "output"}], + } + ) + content_hash = definition_hash(graph, None) + with self.database.session() as session: + pipeline = DataflowPipeline( + id="pipeline-1", + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + definition_kind="flow", + name="Audience", + status="active", + current_revision=1, + allow_run=True, + metadata_={}, + ) + revision = DataflowPipelineRevision( + id="revision-1", + tenant_id="tenant-1", + pipeline_id=pipeline.id, + revision=1, + graph=graph.model_dump(mode="json"), + sql_text=None, + editor_mode="graph", + content_hash=content_hash, + ) + session.add_all((pipeline, revision)) + session.flush() + + provider = SqlDataflowDatasetOutputProvider() + descriptors = provider.list_outputs(session, principal()) + result = provider.read_output( + session, + principal(), + request=DataflowDatasetRequest( + pipeline_ref=pipeline.id, + revision=1, + expected_definition_hash=content_hash, + ), + ) + + self.assertEqual(1, len(descriptors)) + self.assertEqual(2, result.total_rows) + self.assertEqual("one", result.rows[0]["recipient_key"]) + self.assertEqual(64, len(result.output_hash)) + + replay = provider.read_output( + session, + principal(), + request=DataflowDatasetRequest( + pipeline_ref=pipeline.id, + revision=1, + expected_definition_hash=content_hash, + expected_source_fingerprints=result.source_fingerprints, + ), + ) + self.assertEqual(result.output_hash, replay.output_hash) + + with self.assertRaises(DataflowRunConflictError): + provider.read_output( + session, + principal(), + request=DataflowDatasetRequest( + pipeline_ref=pipeline.id, + revision=1, + expected_definition_hash="wrong", + ), + ) + with self.assertRaises(DataflowRunConflictError): + provider.read_output( + session, + principal(), + request=DataflowDatasetRequest( + pipeline_ref=pipeline.id, + revision=1, + expected_definition_hash=content_hash, + expected_source_fingerprints=( + {"node_id": "source", "fingerprint": "changed"}, + ), + ), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_golden_flows.py b/tests/test_golden_flows.py index 6218406..d6414ce 100644 --- a/tests/test_golden_flows.py +++ b/tests/test_golden_flows.py @@ -1,12 +1,19 @@ from __future__ import annotations import json +import hashlib import unittest from pathlib import Path -from govoplan_dataflow.backend.executor import execute_preview -from govoplan_dataflow.backend.graph import validate_graph +from govoplan_dataflow.backend.executor import ( + PipelineExecutionError, + ResolvedSource, + execute_preview, +) +from govoplan_dataflow.backend.expressions import ExpressionError, parse_expression +from govoplan_dataflow.backend.graph import definition_hash, validate_graph from govoplan_dataflow.backend.schemas import PipelineGraph +from govoplan_dataflow.backend.subflows import substitute_parameters FIXTURES = Path(__file__).parents[1] / "fixtures" / "golden" @@ -21,13 +28,18 @@ class GoldenFlowTests(unittest.TestCase): for directory in directories: with self.subTest(flow=directory.name): manifest = _json(directory / "manifest.json") - graph_payload = _json(directory / str(manifest["graph"])) + raw_graph_payload = _json(directory / str(manifest["graph"])) + graph_payload = json.loads(json.dumps(raw_graph_payload)) for item in graph_payload["nodes"]: fixture = item.get("config", {}).get("fixture") - if fixture: + if fixture and item["type"] == "source.inline": item["config"]["rows"] = _json( directory / "inputs" / fixture ) + graph_payload = substitute_parameters( + graph_payload, + dict(manifest.get("parameters") or {}), + ) graph = PipelineGraph.model_validate(graph_payload) diagnostics = validate_graph(graph) self.assertEqual( @@ -39,17 +51,147 @@ class GoldenFlowTests(unittest.TestCase): ], ) - result = execute_preview(graph, row_limit=500) + resolver = _source_resolver(directory, manifest) + result = execute_preview( + graph, + row_limit=500, + source_resolver=resolver, + ) self.assertEqual( _json(directory / str(manifest["expected_output"])), result.rows, ) + if "expected_definition_hash" in manifest: + self.assertEqual( + manifest["expected_definition_hash"], + definition_hash(PipelineGraph.model_validate(raw_graph_payload)), + ) + if "expected_execution_graph_hash" in manifest: + self.assertEqual( + manifest["expected_execution_graph_hash"], + definition_hash(graph), + ) + if "expected_output_hash" in manifest: + self.assertEqual( + manifest["expected_output_hash"], + _rows_hash(result.rows), + ) + if "expected_source_fingerprints" in manifest: + self.assertEqual( + manifest["expected_source_fingerprints"], + result.source_fingerprints, + ) + full_result = execute_preview( + graph, + row_limit=5_000, + source_resolver=resolver, + ) + self.assertEqual(result.rows, full_result.rows) + self.assertFalse(full_result.truncated) + + def test_adrema_fixture_rejects_untrusted_sources_and_unsafe_expression(self) -> None: + directory = FIXTURES / "adrema-audience-selection" + manifest = _json(directory / "manifest.json") + graph = PipelineGraph.model_validate( + substitute_parameters( + _json(directory / "graph.json"), + dict(manifest["parameters"]), + ) + ) + drifted_manifest = { + **manifest, + "sources": { + **manifest["sources"], + "snapshot:directory-2026-01": { + **manifest["sources"]["snapshot:directory-2026-01"], + "fingerprint": "sha256:changed-source", + }, + }, + } + + with self.assertRaisesRegex(PipelineExecutionError, "fingerprint"): + execute_preview( + graph, + row_limit=500, + source_resolver=_source_resolver(directory, drifted_manifest), + ) + cross_tenant_manifest = { + **manifest, + "sources": { + **manifest["sources"], + "snapshot:directory-2026-01": { + **manifest["sources"]["snapshot:directory-2026-01"], + "tenant_id": "tenant-other", + }, + }, + } + with self.assertRaisesRegex(PipelineExecutionError, "tenant"): + execute_preview( + graph, + row_limit=500, + source_resolver=_source_resolver(directory, cross_tenant_manifest), + ) + + unknown_source_graph = graph.model_copy(deep=True) + unknown_source_graph.nodes[0].config["source_ref"] = "connector:untrusted" + with self.assertRaisesRegex(PipelineExecutionError, "pinned provider set"): + execute_preview( + unknown_source_graph, + row_limit=500, + source_resolver=_source_resolver(directory, manifest), + ) + with self.assertRaises(ExpressionError): + parse_expression("read_csv('/tmp/private.csv')") def _json(path: Path): return json.loads(path.read_text(encoding="utf-8")) +def _source_resolver(directory: Path, manifest: dict): + sources = dict(manifest.get("sources") or {}) + tenant_id = str(manifest.get("tenant_id") or "") + + def resolve(node, limit: int) -> ResolvedSource: + source_ref = str(node.config.get("source_ref") or "") + if source_ref not in sources: + raise ValueError(f"Fixture source {source_ref!r} is outside the pinned provider set.") + source = sources[source_ref] + if str(source.get("tenant_id") or "") != tenant_id: + raise ValueError( + f"Fixture source {source_ref!r} belongs to a different tenant." + ) + fingerprint = str(source["fingerprint"]) + expected = str(node.config.get("expected_fingerprint") or "") + if expected and expected != fingerprint: + raise ValueError( + f"Fixture source fingerprint changed for {source_ref!r}." + ) + rows = _json(directory / str(source["fixture"])) + bounded = rows[:limit] + return ResolvedSource( + rows=tuple(dict(item) for item in bounded), + source_ref=source_ref, + provider=str(source["provider"]), + fingerprint=fingerprint, + total_rows=len(rows), + truncated=len(bounded) < len(rows), + ) + + return resolve + + +def _rows_hash(rows: object) -> str: + return hashlib.sha256( + json.dumps( + rows, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + ).hexdigest() + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 31f0f36..ae083ce 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -3,6 +3,7 @@ from __future__ import annotations import unittest from govoplan_core.core.dataflows import ( + CAPABILITY_DATAFLOW_DATASET_OUTPUT, CAPABILITY_DATAFLOW_RUN_LIFECYCLE, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, ) @@ -31,6 +32,10 @@ class DataflowManifestTests(unittest.TestCase): CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, manifest.capability_factories, ) + self.assertIn( + CAPABILITY_DATAFLOW_DATASET_OUTPUT, + manifest.capability_factories, + ) self.assertEqual( { "dataflow.pipeline_catalog",