124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
"""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
|