"""Tests para build_join_graph.""" from build_join_graph import build_join_graph def _star_fks(): """Esquema en estrella: orders apunta a customers y a products.""" return [ { "from_table": "orders", "from_col": "customer_id", "to_table": "customers", "to_col": "id", "inclusion": 1.0, "cardinality": "many-to-one", }, { "from_table": "orders", "from_col": "product_id", "to_table": "products", "to_col": "id", "inclusion": 0.98, "cardinality": "many-to-one", }, ] def test_star_schema_roles_and_hub(): g = build_join_graph(_star_fks()) nodes = {n["table"]: n for n in g["nodes"]} assert nodes["orders"]["role"] == "fact" assert nodes["orders"]["out_degree"] == 2 assert nodes["orders"]["in_degree"] == 0 assert nodes["customers"]["role"] == "dimension" assert nodes["customers"]["in_degree"] == 1 assert nodes["customers"]["out_degree"] == 0 assert nodes["products"]["role"] == "dimension" # orders es el hub (mayor out_degree). assert g["hubs"][0] == "orders" def test_two_edges_built(): g = build_join_graph(_star_fks()) assert len(g["edges"]) == 2 pairs = {(e["from_table"], e["to_table"]) for e in g["edges"]} assert pairs == {("orders", "customers"), ("orders", "products")} def test_mermaid_contains_tables_and_arrows(): g = build_join_graph(_star_fks()) m = g["mermaid"] assert "orders" in m assert "customers" in m assert "products" in m assert "-->" in m # Etiqueta de columnas en la arista. assert "customer_id->id" in m def test_bridge_role(): # order_items apunta a orders y products, y nadie le apunta -> fact en este # subgrafo. Para forzar bridge, hacemos que reciba tambien una FK. fks = [ {"from_table": "shipments", "from_col": "order_item_id", "to_table": "order_items", "to_col": "id", "inclusion": 1.0, "cardinality": "many-to-one"}, {"from_table": "order_items", "from_col": "product_id", "to_table": "products", "to_col": "id", "inclusion": 1.0, "cardinality": "many-to-one"}, ] g = build_join_graph(fks) nodes = {n["table"]: n for n in g["nodes"]} assert nodes["order_items"]["role"] == "bridge" assert nodes["order_items"]["in_degree"] == 1 assert nodes["order_items"]["out_degree"] == 1 def test_standalone_node_from_tables_list(): g = build_join_graph(_star_fks(), tables=["orders", "customers", "products", "audit_log"]) nodes = {n["table"]: n for n in g["nodes"]} assert "audit_log" in nodes assert nodes["audit_log"]["role"] == "standalone" assert nodes["audit_log"]["out_degree"] == 0 assert nodes["audit_log"]["in_degree"] == 0 # El nodo aislado aparece declarado en el mermaid. assert "audit_log" in g["mermaid"] def test_empty_list_does_not_crash(): g = build_join_graph([]) assert g["nodes"] == [] assert g["edges"] == [] assert g["hubs"] == [] assert g["mermaid"].startswith("graph LR") def test_none_input_does_not_crash(): g = build_join_graph(None) assert g["edges"] == [] assert "graph LR" in g["mermaid"] def test_malformed_entries_skipped(): fks = [ {"from_table": "a", "from_col": "x", "to_table": "b", "to_col": "y"}, {"from_table": "a"}, # falta to_table -> se ignora "not a dict", # no es dict -> se ignora {"to_table": "b"}, # falta from_table -> se ignora ] g = build_join_graph(fks) assert len(g["edges"]) == 1 assert g["edges"][0]["from_table"] == "a" def test_does_not_mutate_input(): fks = _star_fks() snapshot = [dict(fk) for fk in fks] build_join_graph(fks) assert fks == snapshot