From fc4d14bce0d22e9685d93367bdcdfb42a2c4ed3c Mon Sep 17 00:00:00 2001 From: Gaurav Bajpai Date: Thu, 23 Apr 2026 22:46:09 +0530 Subject: [PATCH] test(bdt): unit tests + adversarial fixtures @W-22196528@ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the full test suite for bdt_analyze.py: 92 unit tests backed by 9 JSON fixtures (4 normative + 5 adversarial). All tests stdlib-only, runnable via `python3 -m unittest tests.test_bdt_analyze`. tests/test_bdt_analyze.py coverage: DAG primitives: - roots() / sinks() / topo_order() on minimal + branching graphs. - upstream() / downstream() reachability correctness. - Cycle tolerance: topo_order + traversals do not infinite-loop or raise when the graph contains a cycle; cycle is reported via broken_references-adjacent signals. - Broken-reference tolerance: nodes referencing non-existent upstreams are handled gracefully; broken_references() enumerates them. CLI subcommands (happy-path + error-path for each): - summary, sources, outputs, stages, nodes, node, lineage, field-trace, formula, definitions — each exercised in both human-readable and --json modes. - Exit-code discipline: unknown node -> exit 2, malformed input -> exit 3, success -> exit 0. - Unknown-action graceful degradation: nodes whose action isn't in the catalog still appear in summary/nodes output with a generic label rather than crashing. - Output size budgets: default-mode outputs are asserted to stay within configured character caps. Field discovery: - fields_produced() / fields_consumed() / _scrape_field_refs() heuristics tested against both clean and noisy formula bodies. - TestFieldTraceFocus — narrowing tests preventing false-positive substring hits in formula text. Dual-shape input: - TestInputShapeDetection — editor export vs. Connect API single-definition auto-detection. - TestMultiDefinition — Connect API multi-definition payload: definitions subcommand, --definition N routing, out-of-range index error handling. Fixtures (tests/fixtures/): - minimal.json — smallest valid editor-export BDT. - window_and_aggregate.json — window + aggregate composition. - api_input_single.json — Connect API single-definition payload. - api_input_multi.json — Connect API multi-definition payload. Adversarial fixtures (tests/fixtures/adversarial/): - cycle.json — graph with a cycle; parser must not hang. - broken_ref.json — node referencing a non-existent upstream. - unknown_action.json — node with an action not in the catalog. - no_ui.json — BDT missing the UI layer entirely. - empty_nodes.json — valid envelope but zero nodes. Suite status: Ran 92 tests, OK. @W-22196528@ --- .../fixtures/adversarial/broken_ref.json | 7 + .../tests/fixtures/adversarial/cycle.json | 8 + .../fixtures/adversarial/empty_nodes.json | 1 + .../tests/fixtures/adversarial/no_ui.json | 7 + .../fixtures/adversarial/unknown_action.json | 7 + .../tests/fixtures/api_input_multi.json | 71 ++ .../tests/fixtures/api_input_single.json | 38 + .../tests/fixtures/minimal.json | 57 ++ .../tests/fixtures/window_and_aggregate.json | 61 ++ .../tests/test_bdt_analyze.py | 811 ++++++++++++++++++ 10 files changed, 1068 insertions(+) create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/adversarial/broken_ref.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/adversarial/cycle.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/adversarial/empty_nodes.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/adversarial/no_ui.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/adversarial/unknown_action.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/api_input_multi.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/api_input_single.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/minimal.json create mode 100644 skills/explaining-batch-data-transform/tests/fixtures/window_and_aggregate.json create mode 100644 skills/explaining-batch-data-transform/tests/test_bdt_analyze.py diff --git a/skills/explaining-batch-data-transform/tests/fixtures/adversarial/broken_ref.json b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/broken_ref.json new file mode 100644 index 0000000..3904812 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/broken_ref.json @@ -0,0 +1,7 @@ +{ + "version": "66.0", + "nodes": { + "LOAD0": {"action": "load", "sources": [], "parameters": {}}, + "JOIN0": {"action": "join", "sources": ["LOAD0", "NO_SUCH_NODE"], "parameters": {}} + } +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/adversarial/cycle.json b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/cycle.json new file mode 100644 index 0000000..8fb1c97 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/cycle.json @@ -0,0 +1,8 @@ +{ + "version": "66.0", + "nodes": { + "A": {"action": "formula", "sources": ["C"], "parameters": {}}, + "B": {"action": "formula", "sources": ["A"], "parameters": {}}, + "C": {"action": "formula", "sources": ["B"], "parameters": {}} + } +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/adversarial/empty_nodes.json b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/empty_nodes.json new file mode 100644 index 0000000..488d122 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/empty_nodes.json @@ -0,0 +1 @@ +{"version": "66.0", "nodes": {}} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/adversarial/no_ui.json b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/no_ui.json new file mode 100644 index 0000000..7d86418 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/no_ui.json @@ -0,0 +1,7 @@ +{ + "version": "66.0", + "nodes": { + "LOAD0": {"action": "load", "sources": [], "parameters": {}}, + "OUT0": {"action": "outputD360", "sources": ["LOAD0"], "parameters": {}} + } +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/adversarial/unknown_action.json b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/unknown_action.json new file mode 100644 index 0000000..644bd71 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/adversarial/unknown_action.json @@ -0,0 +1,7 @@ +{ + "version": "99.0", + "nodes": { + "LOAD0": {"action": "load", "sources": [], "parameters": {}}, + "MYSTERY0": {"action": "someFutureAction", "sources": ["LOAD0"], "parameters": {"foo": "bar"}} + } +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/api_input_multi.json b/skills/explaining-batch-data-transform/tests/fixtures/api_input_multi.json new file mode 100644 index 0000000..b6b38ac --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/api_input_multi.json @@ -0,0 +1,71 @@ +{ + "name": "TestMultiTransform", + "label": "Test Multi Transform", + "description": "Wraps two BDTs in the Connect API multi-definitions shape", + "type": "BATCH", + "dataSpaceName": "default", + "definitions": [ + { + "version": "66.0", + "type": "STL", + "nodes": { + "LOAD_A": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__A__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "OUTPUT_A": { + "action": "outputD360", + "sources": ["LOAD_A"], + "parameters": { + "name": "A_Out__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [{"sourceField": "ssot__Id__c", "targetField": "ssot__Id__c"}] + } + } + }, + "ui": { + "nodes": { + "LOAD_A": {"label": "Load A", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "OUTPUT_A": {"label": "Output A", "type": "OUTPUT", "top": 100, "left": 260} + } + } + }, + { + "version": "66.0", + "type": "STL", + "nodes": { + "LOAD_B": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__B__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "OUTPUT_B": { + "action": "outputD360", + "sources": ["LOAD_B"], + "parameters": { + "name": "B_Out__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [{"sourceField": "ssot__Id__c", "targetField": "ssot__Id__c"}] + } + } + }, + "ui": { + "nodes": { + "LOAD_B": {"label": "Load B", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "OUTPUT_B": {"label": "Output B", "type": "OUTPUT", "top": 100, "left": 260} + } + } + } + ] +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/api_input_single.json b/skills/explaining-batch-data-transform/tests/fixtures/api_input_single.json new file mode 100644 index 0000000..9332ea3 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/api_input_single.json @@ -0,0 +1,38 @@ +{ + "name": "TestTransform", + "label": "Test Transform", + "description": "Wraps minimal BDT in the Connect API single-definition shape", + "type": "BATCH", + "dataSpaceName": "default", + "definition": { + "version": "66.0", + "type": "STL", + "nodes": { + "LOAD_X": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__X__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "OUTPUT_X": { + "action": "outputD360", + "sources": ["LOAD_X"], + "parameters": { + "name": "X_Out__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [{"sourceField": "ssot__Id__c", "targetField": "ssot__Id__c"}] + } + } + }, + "ui": { + "nodes": { + "LOAD_X": {"label": "Load X", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "OUTPUT_X": {"label": "Output X", "type": "OUTPUT", "top": 100, "left": 260} + } + } + } +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/minimal.json b/skills/explaining-batch-data-transform/tests/fixtures/minimal.json new file mode 100644 index 0000000..c33c8b3 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/minimal.json @@ -0,0 +1,57 @@ +{ + "version": "66.0", + "nodes": { + "LOAD_DATASET0": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__Account__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__Name__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "FORMULA0": { + "action": "formula", + "sources": ["LOAD_DATASET0"], + "parameters": { + "expressionType": "SQL", + "fields": [ + { + "name": "AccountNameUpper__c", + "label": "Account Name Upper", + "formulaExpression": "upper(ssot__Name__c)", + "type": "TEXT", + "businessType": "Text", + "precision": 255, + "defaultValue": "" + } + ] + } + }, + "OUTPUT0": { + "action": "outputD360", + "sources": ["FORMULA0"], + "parameters": { + "name": "Account_Upper__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [ + {"sourceField": "ssot__Id__c", "targetField": "ssot__Id__c"}, + {"sourceField": "AccountNameUpper__c", "targetField": "Name_Upper__c"} + ] + } + } + }, + "ui": { + "nodes": { + "LOAD_DATASET0": {"label": "Account", "type": "LOAD_DATASET", "top": 100, "left": 100}, + "FORMULA0": {"label": "Uppercase Name", "type": "FORMULA", "top": 100, "left": 240}, + "OUTPUT0": {"label": "Account Upper", "type": "OUTPUT", "top": 100, "left": 380} + }, + "connectors": [ + {"source": "LOAD_DATASET0", "target": "FORMULA0"}, + {"source": "FORMULA0", "target": "OUTPUT0"} + ], + "hiddenColumns": [] + } +} diff --git a/skills/explaining-batch-data-transform/tests/fixtures/window_and_aggregate.json b/skills/explaining-batch-data-transform/tests/fixtures/window_and_aggregate.json new file mode 100644 index 0000000..eb060da --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/fixtures/window_and_aggregate.json @@ -0,0 +1,61 @@ +{ + "version": "66.0", + "nodes": { + "LOAD_ORDERS": { + "action": "load", + "sources": [], + "parameters": { + "dataset": {"name": "ssot__SalesOrder__dlm", "type": "dataModelObject"}, + "fields": ["ssot__Id__c", "ssot__AccountId__c", "ssot__CreatedDate__c", "ssot__GrandTotalAmount__c"], + "sampleDetails": {"type": "TopN", "sortBy": []} + } + }, + "RANK_ORDERS": { + "action": "computeRelative", + "sources": ["LOAD_ORDERS"], + "parameters": { + "partitionBy": ["ssot__AccountId__c"], + "orderBy": [{"fieldName": "ssot__CreatedDate__c", "direction": "ASC"}], + "expressionType": "SQL", + "fields": [ + { + "name": "OrderRank__c", + "label": "Order Rank", + "formulaExpression": "row_number()", + "type": "NUMBER", + "businessType": "Number", + "precision": 18, + "scale": 0, + "defaultValue": "" + } + ] + } + }, + "AGG_BY_ACCOUNT": { + "action": "aggregate", + "sources": ["RANK_ORDERS"], + "parameters": { + "groupings": ["ssot__AccountId__c"], + "aggregations": [ + {"action": "SUM", "name": "TotalAmount__c", "source": "ssot__GrandTotalAmount__c"}, + {"action": "COUNT", "name": "OrderCount__c", "source": "ssot__Id__c"} + ], + "nodeType": "STANDARD" + } + }, + "OUTPUT_SUMMARY": { + "action": "outputD360", + "sources": ["AGG_BY_ACCOUNT"], + "parameters": { + "name": "Account_Summary__dlm", + "type": "dataModelObject", + "writeMode": "OVERWRITE", + "fieldsMappings": [ + {"sourceField": "ssot__AccountId__c", "targetField": "AccountId__c"}, + {"sourceField": "TotalAmount__c", "targetField": "TotalAmount__c"}, + {"sourceField": "OrderCount__c", "targetField": "OrderCount__c"} + ] + } + } + } +} diff --git a/skills/explaining-batch-data-transform/tests/test_bdt_analyze.py b/skills/explaining-batch-data-transform/tests/test_bdt_analyze.py new file mode 100644 index 0000000..23df942 --- /dev/null +++ b/skills/explaining-batch-data-transform/tests/test_bdt_analyze.py @@ -0,0 +1,811 @@ +"""Unit tests for bdt_analyze.py. + +Run: + cd afv-library/skills/explaining-batch-data-transform + python -m unittest tests.test_bdt_analyze -v +""" +import argparse +import contextlib +import io +import json +import pathlib +import sys +import unittest + +# Add scripts/ to path so we can import the module +SKILL_DIR = pathlib.Path(__file__).resolve().parent.parent +sys.path.insert(0, str(SKILL_DIR / "scripts")) + +FIXTURES = pathlib.Path(__file__).resolve().parent / "fixtures" + +import bdt_analyze # noqa: E402 + + +class TestLoad(unittest.TestCase): + """Loading a valid BDT JSON.""" + + def test_load_minimal(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + self.assertEqual(bdt.version, "66.0") + self.assertEqual(len(bdt.nodes), 3) + self.assertIn("LOAD_DATASET0", bdt.nodes) + self.assertIn("FORMULA0", bdt.nodes) + self.assertIn("OUTPUT0", bdt.nodes) + + +class TestBadInput(unittest.TestCase): + """Adversarial / malformed inputs. Raise BdtInputError.""" + + def test_invalid_json_raises(self): + p = FIXTURES / "_tmp_invalid.json" + p.write_text("{not valid json") + try: + with self.assertRaises(bdt_analyze.BdtInputError) as cm: + bdt_analyze.DataTransform.from_path(p) + self.assertIn("Invalid JSON", str(cm.exception)) + finally: + p.unlink(missing_ok=True) + + def test_missing_nodes_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError): + bdt_analyze.DataTransform.from_dict({"version": "66.0"}) + + def test_nodes_not_object_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError): + bdt_analyze.DataTransform.from_dict({"nodes": "not-an-object"}) + + def test_node_missing_action_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError) as cm: + bdt_analyze.DataTransform.from_dict({ + "nodes": {"X": {"sources": []}} + }) + self.assertIn("missing string 'action'", str(cm.exception)) + + def test_node_sources_not_list_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError): + bdt_analyze.DataTransform.from_dict({ + "nodes": {"X": {"action": "load", "sources": "not-a-list"}} + }) + + def test_nonexistent_file_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError): + bdt_analyze.DataTransform.from_path(FIXTURES / "_does_not_exist.json") + + +class TestEmptyNodes(unittest.TestCase): + """`nodes: {}` is valid — no graph, but don't crash.""" + + def test_empty_nodes_parses(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "empty_nodes.json") + self.assertEqual(bdt.nodes, {}) + + +class TestGraph(unittest.TestCase): + """Graph primitives — roots, sinks, topo order.""" + + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + + def test_roots(self): + roots = self.bdt.roots() + self.assertEqual(roots, ["LOAD_DATASET0"]) + + def test_sinks(self): + sinks = self.bdt.sinks() + self.assertEqual(sinks, ["OUTPUT0"]) + + def test_topo_order(self): + order = self.bdt.topo_order() + self.assertEqual(order, ["LOAD_DATASET0", "FORMULA0", "OUTPUT0"]) + + def test_topo_order_is_valid(self): + """Every node appears after all its sources.""" + order = self.bdt.topo_order() + position = {name: i for i, name in enumerate(order)} + for n in self.bdt.nodes.values(): + for s in n.sources: + self.assertLess(position[s], position[n.name], + f"{s} should come before {n.name}") + + +class TestCycle(unittest.TestCase): + """Cycle detection surfaces a clear error.""" + + def test_cycle_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError) as cm: + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "cycle.json") + bdt.topo_order() + msg = str(cm.exception) + self.assertIn("Cycle", msg) + # All three nodes are stuck + self.assertIn("'A'", msg) + self.assertIn("'B'", msg) + self.assertIn("'C'", msg) + + +class TestTraversal(unittest.TestCase): + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + + def test_upstream_of_output(self): + # OUTPUT0 ← FORMULA0 ← LOAD_DATASET0 + up = self.bdt.upstream("OUTPUT0") + self.assertEqual(up, ["LOAD_DATASET0", "FORMULA0"]) # topo order + + def test_upstream_of_root_is_empty(self): + self.assertEqual(self.bdt.upstream("LOAD_DATASET0"), []) + + def test_upstream_unknown_node_raises(self): + with self.assertRaises(bdt_analyze.BdtNotFoundError): + self.bdt.upstream("NO_SUCH_NODE") + + def test_downstream_of_root(self): + # LOAD_DATASET0 → FORMULA0 → OUTPUT0 + down = self.bdt.downstream("LOAD_DATASET0") + self.assertEqual(down, ["FORMULA0", "OUTPUT0"]) + + def test_downstream_of_sink_is_empty(self): + self.assertEqual(self.bdt.downstream("OUTPUT0"), []) + + +class TestBrokenReferences(unittest.TestCase): + def test_broken_references_reported(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "broken_ref.json") + broken = bdt.broken_references() + # One entry: (referring_node, missing_source) + self.assertEqual(broken, [("JOIN0", "NO_SUCH_NODE")]) + + def test_no_broken_references_on_clean_bdt(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + self.assertEqual(bdt.broken_references(), []) + + def test_topo_raises_on_broken_ref(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "broken_ref.json") + with self.assertRaises(bdt_analyze.BdtInputError): + bdt.topo_order() + + +class TestResilience(unittest.TestCase): + """Skill must never crash on schema evolution or missing ui section.""" + + def test_unknown_action_loads_without_crash(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "unknown_action.json") + self.assertEqual(bdt.nodes["MYSTERY0"].action, "someFutureAction") + # Parameters preserved verbatim + self.assertEqual(bdt.nodes["MYSTERY0"].parameters, {"foo": "bar"}) + # Graph operations still work + self.assertEqual(bdt.topo_order(), ["LOAD0", "MYSTERY0"]) + + def test_no_ui_section_loads_without_crash(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "no_ui.json") + # display_name falls back to node key when no ui label + self.assertEqual(bdt.nodes["LOAD0"].display_name, "LOAD0") + + def test_display_name_uses_label_when_present(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + self.assertEqual(bdt.nodes["LOAD_DATASET0"].display_name, "Account (LOAD_DATASET0)") + + +class TestSummary(unittest.TestCase): + def test_summary_markdown_has_key_sections(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_summary(bdt, args) + # Top-line shape + self.assertIn("# BDT Summary", out) + self.assertIn("Version: 66.0", out) + self.assertIn("Total nodes: 3", out) + # Action counts + self.assertIn("load: 1", out) + self.assertIn("formula: 1", out) + self.assertIn("outputD360: 1", out) + # Sources table includes the load's dataset + self.assertIn("ssot__Account__dlm", out) + # Outputs table includes the output's target + self.assertIn("Account_Upper__dlm", out) + + def test_summary_json_mode(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=True) + out = bdt_analyze.cmd_summary(bdt, args) + parsed = json.loads(out) # must be valid JSON + self.assertEqual(parsed["version"], "66.0") + self.assertEqual(parsed["total_nodes"], 3) + self.assertEqual(parsed["action_counts"]["load"], 1) + + def test_summary_empty_nodes(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "empty_nodes.json") + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_summary(bdt, args) + self.assertIn("Total nodes: 0", out) + + +class TestSources(unittest.TestCase): + def test_sources_markdown(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_sources(bdt, args) + self.assertIn("# Sources", out) + self.assertIn("LOAD_DATASET0", out) + self.assertIn("ssot__Account__dlm", out) + # Fields are listed + self.assertIn("ssot__Id__c", out) + self.assertIn("ssot__Name__c", out) + + def test_sources_json(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=True) + out = bdt_analyze.cmd_sources(bdt, args) + data = json.loads(out) + self.assertEqual(len(data), 1) + self.assertEqual(data[0]["name"], "LOAD_DATASET0") + self.assertIn("ssot__Id__c", data[0]["fields"]) + + +class TestOutputs(unittest.TestCase): + def test_outputs_markdown(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_outputs(bdt, args) + self.assertIn("# Outputs", out) + self.assertIn("OUTPUT0", out) + self.assertIn("Account_Upper__dlm", out) + self.assertIn("OVERWRITE", out) + # Mapping table + self.assertIn("ssot__Id__c", out) + self.assertIn("Name_Upper__c", out) + + def test_outputs_json(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=True) + data = json.loads(bdt_analyze.cmd_outputs(bdt, args)) + self.assertEqual(len(data), 1) + self.assertEqual(data[0]["target_name"], "Account_Upper__dlm") + self.assertEqual(data[0]["write_mode"], "OVERWRITE") + + +class TestStages(unittest.TestCase): + def test_stages_follows_topo_order(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_stages(bdt, args) + self.assertIn("# Stages", out) + # Ordering: LOAD_DATASET0 must come before FORMULA0 which must come before OUTPUT0 + i_load = out.index("LOAD_DATASET0") + i_formula = out.index("FORMULA0") + i_output = out.index("OUTPUT0") + self.assertLess(i_load, i_formula) + self.assertLess(i_formula, i_output) + # Hints present + self.assertIn("load", out.lower()) + self.assertIn("formula", out.lower()) + self.assertIn("outputD360", out) + + def test_stages_json(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=True) + data = json.loads(bdt_analyze.cmd_stages(bdt, args)) + self.assertEqual([n["name"] for n in data], + ["LOAD_DATASET0", "FORMULA0", "OUTPUT0"]) + + +class TestNodes(unittest.TestCase): + def test_nodes_table_present(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, limit=0) + out = bdt_analyze.cmd_nodes(bdt, args) + self.assertIn("| Node |", out) + self.assertIn("LOAD_DATASET0", out) + self.assertIn("FORMULA0", out) + self.assertIn("OUTPUT0", out) + + def test_nodes_limit(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, limit=1) + out = bdt_analyze.cmd_nodes(bdt, args) + # Only the first node appears (LOAD_DATASET0 in topo order) + self.assertIn("LOAD_DATASET0", out) + self.assertNotIn("OUTPUT0", out) + self.assertIn("limited to 1", out.lower()) + + +class TestNodeDetail(unittest.TestCase): + def test_node_detail_markdown(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, node_name="FORMULA0") + out = bdt_analyze.cmd_node(bdt, args) + self.assertIn("# FORMULA0", out) + self.assertIn("formula", out) + self.assertIn("upper(ssot__Name__c)", out) + # Sources and consumers listed + self.assertIn("LOAD_DATASET0", out) + self.assertIn("OUTPUT0", out) + + def test_node_detail_unknown_raises_not_found(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, node_name="NOPE") + with self.assertRaises(bdt_analyze.BdtNotFoundError): + bdt_analyze.cmd_node(bdt, args) + + def test_node_cli_unknown_exits_2(self): + """End-to-end: `bdt_analyze.py node NOPE` exits 2.""" + test = TestCliDispatch() + code, out, err = test._run(["node", str(FIXTURES / "minimal.json"), "NOPE"]) + self.assertEqual(code, 2) + self.assertIn("No node named", err) + + +class TestCliDispatch(unittest.TestCase): + """Top-level CLI: argparse routing + exit codes.""" + + def _run(self, argv): + """Run bdt_analyze.main with argv and capture (exit_code, stdout, stderr).""" + out, err = io.StringIO(), io.StringIO() + code = None + try: + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = bdt_analyze.main(argv) + except SystemExit as e: + code = e.code + return code, out.getvalue(), err.getvalue() + + def test_no_args_prints_help_exit_2(self): + code, out, err = self._run([]) + # argparse exits 2 when no subcommand is given + self.assertEqual(code, 2) + self.assertIn("usage:", (out + err).lower()) + + def test_unknown_subcommand_exit_2(self): + code, out, err = self._run(["nonexistent", str(FIXTURES / "minimal.json")]) + self.assertEqual(code, 2) + + def test_missing_file_exit_3(self): + code, out, err = self._run(["summary", str(FIXTURES / "_does_not_exist.json")]) + self.assertEqual(code, 3) + self.assertIn("cannot read", err.lower()) + + +class TestOutputSizeBudget(unittest.TestCase): + """Outputs must stay within the 5 KB per-subcommand budget on a tiny BDT.""" + + SIZE_LIMIT = 5 * 1024 # 5 KB — per spec §3 + + def _run(self, argv): + test = TestCliDispatch() + return test._run(argv) + + def test_summary_size(self): + code, out, err = self._run(["summary", str(FIXTURES / "minimal.json")]) + self.assertEqual(code, 0) + self.assertLess(len(out), self.SIZE_LIMIT) + + def test_stages_size(self): + code, out, err = self._run(["stages", str(FIXTURES / "minimal.json")]) + self.assertEqual(code, 0) + self.assertLess(len(out), self.SIZE_LIMIT) + + def test_nodes_size(self): + code, out, err = self._run(["nodes", str(FIXTURES / "minimal.json")]) + self.assertEqual(code, 0) + self.assertLess(len(out), self.SIZE_LIMIT) + + def test_node_detail_size(self): + code, out, err = self._run(["node", str(FIXTURES / "minimal.json"), "FORMULA0"]) + self.assertEqual(code, 0) + self.assertLess(len(out), 2 * 1024) # node detail target: ≤ 2 KB + + def test_lineage_size(self): + test = TestCliDispatch() + code, out, err = test._run(["lineage", str(FIXTURES / "minimal.json"), "OUTPUT0"]) + self.assertEqual(code, 0) + self.assertLess(len(out), 2 * 1024) # L1 target: ≤ 2 KB + + def test_field_trace_size(self): + test = TestCliDispatch() + code, out, err = test._run(["field-trace", str(FIXTURES / "minimal.json"), + "AccountNameUpper__c"]) + self.assertEqual(code, 0) + self.assertLess(len(out), 1 * 1024) # L3 target: ≤ 1 KB + + def test_formula_size(self): + test = TestCliDispatch() + code, out, err = test._run(["formula", str(FIXTURES / "minimal.json"), "FORMULA0"]) + self.assertEqual(code, 0) + self.assertLess(len(out), 1 * 1024) # I1 target: ≤ 1 KB + + +class TestTraversalResilience(unittest.TestCase): + """Upstream/downstream must not crash on BDTs with broken refs or cycles, + as long as the queried node's own walk is well-formed.""" + + def test_upstream_works_with_unrelated_broken_ref(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "broken_ref.json") + # JOIN0's second source NO_SUCH_NODE is broken; LOAD0 is clean. + # upstream('JOIN0') should return the clean ancestors, not crash. + up = bdt.upstream("JOIN0") + self.assertIn("LOAD0", up) + + def test_downstream_works_when_unrelated_nodes_have_broken_refs(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "broken_ref.json") + down = bdt.downstream("LOAD0") + self.assertIn("JOIN0", down) + + def test_broken_references_is_sorted(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "adversarial" / "broken_ref.json") + broken = bdt.broken_references() + self.assertEqual(broken, sorted(broken)) + + def test_from_path_on_directory_raises_input_error(self): + """OSError subclasses (IsADirectoryError) should be mapped to BdtInputError.""" + import pathlib + with self.assertRaises(bdt_analyze.BdtInputError) as cm: + bdt_analyze.DataTransform.from_path(FIXTURES) # the fixtures dir itself + self.assertIn("Cannot read", str(cm.exception)) + + +class TestLineage(unittest.TestCase): + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + + def test_lineage_of_output_markdown(self): + args = argparse.Namespace(as_json=False, node_name="OUTPUT0") + out = bdt_analyze.cmd_lineage(self.bdt, args) + self.assertIn("# Lineage of OUTPUT0", out) + # Both upstream nodes named, in topo order + i_load = out.index("LOAD_DATASET0") + i_formula = out.index("FORMULA0") + self.assertLess(i_load, i_formula) + + def test_lineage_of_root_says_root(self): + args = argparse.Namespace(as_json=False, node_name="LOAD_DATASET0") + out = bdt_analyze.cmd_lineage(self.bdt, args) + self.assertIn("is a graph root", out) + + def test_lineage_unknown_raises(self): + args = argparse.Namespace(as_json=False, node_name="NOPE") + with self.assertRaises(bdt_analyze.BdtNotFoundError): + bdt_analyze.cmd_lineage(self.bdt, args) + + def test_lineage_json(self): + args = argparse.Namespace(as_json=True, node_name="OUTPUT0") + data = json.loads(bdt_analyze.cmd_lineage(self.bdt, args)) + self.assertEqual(data["target"], "OUTPUT0") + self.assertEqual(data["upstream"], + [{"name": "LOAD_DATASET0", "action": "load", + "label": "Account", "sources": []}, + {"name": "FORMULA0", "action": "formula", + "label": "Uppercase Name", "sources": ["LOAD_DATASET0"]}]) + + +class TestFieldDiscovery(unittest.TestCase): + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + + def test_load_produces_its_field_list(self): + n = self.bdt.nodes["LOAD_DATASET0"] + produced = bdt_analyze.fields_produced(n) + self.assertIn("ssot__Id__c", produced) + self.assertIn("ssot__Name__c", produced) + + def test_load_consumes_nothing(self): + n = self.bdt.nodes["LOAD_DATASET0"] + self.assertEqual(bdt_analyze.fields_consumed(n), []) + + def test_formula_produces_its_output_field(self): + n = self.bdt.nodes["FORMULA0"] + produced = bdt_analyze.fields_produced(n) + self.assertEqual(produced, ["AccountNameUpper__c"]) + + def test_formula_consumes_fields_mentioned_in_expression(self): + n = self.bdt.nodes["FORMULA0"] + consumed = bdt_analyze.fields_consumed(n) + self.assertIn("ssot__Name__c", consumed) + + def test_output_consumes_source_fields_from_mappings(self): + n = self.bdt.nodes["OUTPUT0"] + consumed = bdt_analyze.fields_consumed(n) + self.assertIn("ssot__Id__c", consumed) + self.assertIn("AccountNameUpper__c", consumed) + + def test_output_produces_target_fields_from_mappings(self): + n = self.bdt.nodes["OUTPUT0"] + produced = bdt_analyze.fields_produced(n) + self.assertIn("ssot__Id__c", produced) + self.assertIn("Name_Upper__c", produced) + + +class TestFieldTrace(unittest.TestCase): + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + + def test_trace_field_defined_in_formula(self): + args = argparse.Namespace(as_json=False, field_name="AccountNameUpper__c") + out = bdt_analyze.cmd_field_trace(self.bdt, args) + self.assertIn("AccountNameUpper__c", out) + # Defined by FORMULA0 + self.assertIn("FORMULA0", out) + # Back-trace reaches the load + self.assertIn("LOAD_DATASET0", out) + # Shows the expression + self.assertIn("upper(ssot__Name__c)", out) + + def test_trace_field_defined_in_load(self): + args = argparse.Namespace(as_json=False, field_name="ssot__Id__c") + out = bdt_analyze.cmd_field_trace(self.bdt, args) + self.assertIn("LOAD_DATASET0", out) + + def test_trace_unknown_field(self): + args = argparse.Namespace(as_json=False, field_name="NOPE__c") + with self.assertRaises(bdt_analyze.BdtNotFoundError) as cm: + bdt_analyze.cmd_field_trace(self.bdt, args) + self.assertIn("NOPE__c", str(cm.exception)) + + def test_trace_json(self): + args = argparse.Namespace(as_json=True, field_name="AccountNameUpper__c") + data = json.loads(bdt_analyze.cmd_field_trace(self.bdt, args)) + self.assertEqual(data["field"], "AccountNameUpper__c") + self.assertTrue(any(e["node"] == "FORMULA0" for e in data["definitions"])) + + +class TestFieldTraceFocus(unittest.TestCase): + """field-trace should narrow upstream deps to the specific field, not + dump all fields consumed by the defining node.""" + + def test_output_node_only_reports_source_of_mapped_field(self): + # In minimal.json OUTPUT0 has 2 field mappings; tracing one targetField + # should list only that mapping's sourceField as the dep, not both. + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=True, field_name="Name_Upper__c") + data = json.loads(bdt_analyze.cmd_field_trace(bdt, args)) + # Find the entry where the definer is OUTPUT0 + output_defs = [d for d in data["definitions"] if d["node"] == "OUTPUT0"] + if output_defs: + # Only AccountNameUpper__c should be listed as an upstream dep, + # NOT ssot__Id__c (which maps to ssot__Id__c target, unrelated to Name_Upper__c). + deps = output_defs[0]["upstream_deps"] + self.assertIn("AccountNameUpper__c", deps) + self.assertNotIn("ssot__Id__c", deps) + + def test_formula_node_reports_only_fields_in_that_expression(self): + # FORMULA0 in minimal.json has one output field AccountNameUpper__c + # with expression upper(ssot__Name__c). Tracing AccountNameUpper__c + # should show exactly ssot__Name__c as a dep. + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=True, field_name="AccountNameUpper__c") + data = json.loads(bdt_analyze.cmd_field_trace(bdt, args)) + formula_defs = [d for d in data["definitions"] if d["node"] == "FORMULA0"] + self.assertEqual(len(formula_defs), 1) + deps = formula_defs[0]["upstream_deps"] + self.assertIn("ssot__Name__c", deps) + + def test_compute_relative_reports_partition_and_order_fields(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "window_and_aggregate.json") + args = argparse.Namespace(as_json=True, field_name="OrderRank__c") + data = json.loads(bdt_analyze.cmd_field_trace(bdt, args)) + rank_defs = [d for d in data["definitions"] if d["node"] == "RANK_ORDERS"] + self.assertEqual(len(rank_defs), 1) + deps = rank_defs[0]["upstream_deps"] + # partitionBy field + self.assertIn("ssot__AccountId__c", deps) + # orderBy field + self.assertIn("ssot__CreatedDate__c", deps) + + def test_aggregate_reports_only_source_of_matching_aggregation(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "window_and_aggregate.json") + args = argparse.Namespace(as_json=True, field_name="TotalAmount__c") + data = json.loads(bdt_analyze.cmd_field_trace(bdt, args)) + agg_defs = [d for d in data["definitions"] if d["node"] == "AGG_BY_ACCOUNT"] + self.assertEqual(len(agg_defs), 1) + deps = agg_defs[0]["upstream_deps"] + # Only the SUM source — not the COUNT source, not the groupings. + self.assertIn("ssot__GrandTotalAmount__c", deps) + self.assertNotIn("ssot__Id__c", deps) + + +class TestWindowAndAggregateTrace(unittest.TestCase): + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "window_and_aggregate.json") + + def test_order_rank_defined_by_compute_relative(self): + args = argparse.Namespace(as_json=False, field_name="OrderRank__c") + out = bdt_analyze.cmd_field_trace(self.bdt, args) + self.assertIn("RANK_ORDERS", out) + self.assertIn("row_number()", out) + # Partition and order are reported as upstream deps + self.assertIn("ssot__AccountId__c", out) + self.assertIn("ssot__CreatedDate__c", out) + + def test_total_amount_traces_through_aggregate(self): + args = argparse.Namespace(as_json=False, field_name="TotalAmount__c") + out = bdt_analyze.cmd_field_trace(self.bdt, args) + # AGG_BY_ACCOUNT produces TotalAmount__c + self.assertIn("AGG_BY_ACCOUNT", out) + self.assertIn("SUM", out) + self.assertIn("ssot__GrandTotalAmount__c", out) + + def test_aggregate_groupings_carried_through(self): + # AccountId__c → ssot__AccountId__c passes from aggregate output through output mapping + args = argparse.Namespace(as_json=False, field_name="AccountId__c") + out = bdt_analyze.cmd_field_trace(self.bdt, args) + self.assertIn("OUTPUT_SUMMARY", out) + self.assertIn("ssot__AccountId__c", out) + + +class TestFormulaSubcommand(unittest.TestCase): + def test_formula_of_formula_node(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, node_name="FORMULA0") + out = bdt_analyze.cmd_formula(bdt, args) + self.assertIn("FORMULA0", out) + self.assertIn("upper(ssot__Name__c)", out) + # Upstream field `ssot__Name__c` is defined by LOAD_DATASET0 + self.assertIn("ssot__Name__c", out) + self.assertIn("LOAD_DATASET0", out) + + def test_formula_of_compute_relative(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "window_and_aggregate.json") + args = argparse.Namespace(as_json=False, node_name="RANK_ORDERS") + out = bdt_analyze.cmd_formula(bdt, args) + self.assertIn("row_number()", out) + self.assertIn("partitionBy", out) + self.assertIn("ssot__AccountId__c", out) + + def test_formula_of_non_formula_node_rejects(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, node_name="LOAD_DATASET0") + with self.assertRaises(bdt_analyze.BdtInputError): + bdt_analyze.cmd_formula(bdt, args) + + def test_formula_unknown_node(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + args = argparse.Namespace(as_json=False, node_name="NOPE") + with self.assertRaises(bdt_analyze.BdtNotFoundError): + bdt_analyze.cmd_formula(bdt, args) + + +class TestInputShapeDetection(unittest.TestCase): + """Parser accepts both the editor-export shape and the Connect API create payload shape.""" + + def test_editor_export_shape_still_works(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + self.assertEqual(len(bdt.nodes), 3) + # No outer wrapper metadata + self.assertIsNone(bdt.name) + self.assertIsNone(bdt.data_transform_type) + # `definitions` is always non-empty; length 1 for editor-export. + self.assertEqual(bdt.definition_count, 1) + + def test_api_input_single_definition(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "api_input_single.json") + self.assertEqual(bdt.name, "TestTransform") + self.assertEqual(bdt.label, "Test Transform") + self.assertEqual(bdt.data_transform_type, "BATCH") + self.assertEqual(bdt.data_space_name, "default") + self.assertEqual(bdt.definition_count, 1) + self.assertEqual(bdt.definition_index, 0) + # Nodes accessible at top of DataTransform + self.assertIn("LOAD_X", bdt.nodes) + self.assertIn("OUTPUT_X", bdt.nodes) + # Version from the definition (not the wrapper) + self.assertEqual(bdt.version, "66.0") + # No warnings for single-definition + self.assertEqual(bdt.warnings(), []) + + def test_api_input_multi_definition(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "api_input_multi.json") + self.assertEqual(bdt.definition_count, 2) + self.assertEqual(bdt.definition_index, 0) + # Descends into first definition by default + self.assertIn("LOAD_A", bdt.nodes) + self.assertNotIn("LOAD_B", bdt.nodes) + # Multi-definition no longer emits a "first-only" warning — the parser + # now exposes every definition and the CLI has a `--definition N` flag. + self.assertEqual(bdt.warnings(), []) + + def test_unknown_shape_raises(self): + with self.assertRaises(bdt_analyze.BdtInputError) as cm: + bdt_analyze.DataTransform.from_dict({"name": "MyTransform", "label": "L"}) + self.assertIn("Expected 'nodes'", str(cm.exception)) + + def test_all_subcommands_work_on_api_input(self): + """summary, stages, lineage, field-trace should all work on the API-wrapped BDT.""" + import io, contextlib + for subcmd, extra_arg in [("summary", None), ("stages", None), ("nodes", None), + ("sources", None), ("outputs", None), + ("lineage", "OUTPUT_X"), ("node", "LOAD_X")]: + argv = [subcmd, str(FIXTURES / "api_input_single.json")] + if extra_arg: argv.append(extra_arg) + out = io.StringIO(); err = io.StringIO() + try: + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = bdt_analyze.main(argv) + except SystemExit as e: + code = e.code + self.assertEqual(code, 0, f"Subcommand {subcmd} failed: {err.getvalue()}") + + +class TestMultiDefinition(unittest.TestCase): + """Parser exposes every definition in a multi-definition payload, not just the first.""" + + def setUp(self): + self.bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "api_input_multi.json") + + def test_definitions_length(self): + self.assertEqual(len(self.bdt.definitions), 2) + self.assertEqual(self.bdt.definition_count, 2) + # Default selected index is 0 + self.assertEqual(self.bdt.definition_index, 0) + + def test_default_selection_is_first(self): + self.assertIn("LOAD_A", self.bdt.nodes) + self.assertNotIn("LOAD_B", self.bdt.nodes) + + def test_select_second_definition(self): + self.bdt.select_definition(1) + self.assertEqual(self.bdt.definition_index, 1) + self.assertIn("LOAD_B", self.bdt.nodes) + self.assertNotIn("LOAD_A", self.bdt.nodes) + + def test_select_out_of_range_raises(self): + with self.assertRaises(bdt_analyze.BdtNotFoundError) as cm: + self.bdt.select_definition(5) + self.assertIn("out of range", str(cm.exception)) + + def test_warnings_no_longer_mentions_first_only(self): + # Previously warned "2 definitions; explains first only". No longer should. + warnings = self.bdt.warnings() + self.assertEqual(warnings, []) + + def test_definitions_subcommand_markdown(self): + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_definitions(self.bdt, args) + self.assertIn("# Definitions", out) + self.assertIn("| 0 |", out) + self.assertIn("| 1 |", out) + + def test_definitions_subcommand_json(self): + args = argparse.Namespace(as_json=True) + data = json.loads(bdt_analyze.cmd_definitions(self.bdt, args)) + self.assertEqual(len(data), 2) + self.assertEqual(data[0]["index"], 0) + self.assertEqual(data[1]["index"], 1) + + def test_definition_flag_routes_summary_to_second(self): + """Full CLI smoke: --definition 1 makes summary operate on the second definition.""" + import io, contextlib + argv = ["summary", "--definition", "1", str(FIXTURES / "api_input_multi.json")] + out, err = io.StringIO(), io.StringIO() + try: + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = bdt_analyze.main(argv) + except SystemExit as e: + code = e.code + self.assertEqual(code, 0, err.getvalue()) + self.assertIn("LOAD_B", out.getvalue()) + self.assertNotIn("LOAD_A", out.getvalue()) + + def test_definition_flag_out_of_range_exits_2(self): + import io, contextlib + argv = ["summary", "--definition", "5", str(FIXTURES / "api_input_multi.json")] + out, err = io.StringIO(), io.StringIO() + try: + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = bdt_analyze.main(argv) + except SystemExit as e: + code = e.code + self.assertEqual(code, 2) + self.assertIn("out of range", err.getvalue()) + + def test_editor_export_has_single_definition(self): + bdt = bdt_analyze.DataTransform.from_path(FIXTURES / "minimal.json") + self.assertEqual(len(bdt.definitions), 1) + self.assertEqual(bdt.definition_count, 1) + # definitions subcommand on a single-definition input still works + args = argparse.Namespace(as_json=False) + out = bdt_analyze.cmd_definitions(bdt, args) + self.assertIn("# Definitions", out) + self.assertIn("| 0 |", out) + + +if __name__ == "__main__": + unittest.main()