mirror of
https://github.com/q3alique/codeflow
synced 2026-06-21 14:06:02 +00:00
q3alique
acc1b4f3e7
Static taint-analysis and visualization tool for source-code security review. Supports deep analysis for Python, JavaScript/TypeScript, Java, Go, and C#, with structural support for all other languages via the generic extractor. Outputs: interactive HTML report, LLM-ready Markdown review document, and optional Burp Suite JSON export. Self-bootstrapping launcher (run.py) requires no virtual environment.
206 lines
7.2 KiB
Python
206 lines
7.2 KiB
Python
"""Task 002 tests: data models."""
|
|
import pytest
|
|
from codeflow.models import Node, NodeType, Edge, EdgeType, CodeGraph, Flow
|
|
from codeflow.models.node import make_node_id, make_module_node_id, new_cluster_id
|
|
|
|
|
|
class TestNode:
|
|
def test_create_node(self):
|
|
n = Node(id="test::func", node_type=NodeType.FUNCTION, label="func")
|
|
assert n.id == "test::func"
|
|
assert n.node_type == NodeType.FUNCTION
|
|
assert n.label == "func"
|
|
|
|
def test_node_defaults(self):
|
|
n = Node(id="x", node_type=NodeType.MODULE, label="x")
|
|
assert n.file_path == ""
|
|
assert n.line_start == 0
|
|
assert n.is_entry_point is False
|
|
assert n.is_tainted is False
|
|
assert n.metadata == {}
|
|
|
|
def test_node_type_values(self):
|
|
assert NodeType.SOURCE.value == "SOURCE"
|
|
assert NodeType.SINK.value == "SINK"
|
|
assert NodeType.ENTRY_POINT.value == "ENTRY_POINT"
|
|
assert NodeType.FUNCTION.value == "FUNCTION"
|
|
assert NodeType.CLASS.value == "CLASS"
|
|
assert NodeType.CLUSTER.value == "CLUSTER"
|
|
assert NodeType.BOUNDARY.value == "BOUNDARY"
|
|
assert NodeType.MODULE.value == "MODULE"
|
|
assert NodeType.ENV_VAR.value == "ENV_VAR"
|
|
|
|
def test_make_node_id(self):
|
|
assert make_node_id("app", "login") == "app::login"
|
|
|
|
def test_make_module_node_id(self):
|
|
assert make_module_node_id("app") == "module::app"
|
|
|
|
def test_new_cluster_id_unique(self):
|
|
ids = {new_cluster_id() for _ in range(100)}
|
|
assert len(ids) == 100
|
|
|
|
def test_node_metadata(self):
|
|
n = Node(id="t::f", node_type=NodeType.FUNCTION, label="f",
|
|
metadata={"param": "value"})
|
|
assert n.metadata["param"] == "value"
|
|
|
|
def test_is_tainted(self):
|
|
n = Node(id="t::src", node_type=NodeType.SOURCE, label="src", is_tainted=True)
|
|
assert n.is_tainted is True
|
|
|
|
|
|
class TestEdge:
|
|
def test_create_edge(self):
|
|
e = Edge(source_id="a", target_id="b", edge_type=EdgeType.CALLS)
|
|
assert e.source_id == "a"
|
|
assert e.target_id == "b"
|
|
assert e.edge_type == EdgeType.CALLS
|
|
|
|
def test_edge_type_values(self):
|
|
assert EdgeType.TAINTED_FLOW.value == "TAINTED_FLOW"
|
|
assert EdgeType.CALLS.value == "CALLS"
|
|
assert EdgeType.DATA_FLOW.value == "DATA_FLOW"
|
|
assert EdgeType.CONTAINS.value == "CONTAINS"
|
|
assert EdgeType.IMPORTS.value == "IMPORTS"
|
|
assert EdgeType.CLUSTER_MEMBER.value == "CLUSTER_MEMBER"
|
|
|
|
def test_edge_defaults(self):
|
|
e = Edge(source_id="a", target_id="b", edge_type=EdgeType.CALLS)
|
|
assert e.metadata == {}
|
|
|
|
|
|
class TestCodeGraph:
|
|
def test_empty_graph(self):
|
|
g = CodeGraph()
|
|
assert g.node_count() == 0
|
|
assert g.edge_count() == 0
|
|
assert g.nodes() == []
|
|
|
|
def test_add_node(self):
|
|
g = CodeGraph()
|
|
n = Node(id="mod::app", node_type=NodeType.MODULE, label="app.py")
|
|
g.add_node(n)
|
|
assert g.node_count() == 1
|
|
assert g.has_node("mod::app")
|
|
|
|
def test_get_node(self):
|
|
g = CodeGraph()
|
|
n = Node(id="f::x", node_type=NodeType.FUNCTION, label="x")
|
|
g.add_node(n)
|
|
retrieved = g.get_node("f::x")
|
|
assert retrieved is not None
|
|
assert retrieved.label == "x"
|
|
|
|
def test_get_nonexistent_node(self):
|
|
g = CodeGraph()
|
|
assert g.get_node("nonexistent") is None
|
|
|
|
def test_add_edge(self):
|
|
g = CodeGraph()
|
|
n1 = Node(id="a", node_type=NodeType.MODULE, label="a")
|
|
n2 = Node(id="b", node_type=NodeType.FUNCTION, label="b")
|
|
g.add_node(n1)
|
|
g.add_node(n2)
|
|
e = Edge(source_id="a", target_id="b", edge_type=EdgeType.CONTAINS)
|
|
g.add_edge(e)
|
|
assert g.edge_count() == 1
|
|
assert g.has_edge("a", "b")
|
|
|
|
def test_edge_only_when_nodes_exist(self):
|
|
g = CodeGraph()
|
|
n1 = Node(id="a", node_type=NodeType.MODULE, label="a")
|
|
g.add_node(n1)
|
|
e = Edge(source_id="a", target_id="nonexistent", edge_type=EdgeType.CALLS)
|
|
g.add_edge(e)
|
|
assert g.edge_count() == 0
|
|
|
|
def test_get_edge(self):
|
|
g = CodeGraph()
|
|
g.add_node(Node(id="a", node_type=NodeType.MODULE, label="a"))
|
|
g.add_node(Node(id="b", node_type=NodeType.FUNCTION, label="b"))
|
|
e = Edge(source_id="a", target_id="b", edge_type=EdgeType.CALLS)
|
|
g.add_edge(e)
|
|
retrieved = g.get_edge("a", "b")
|
|
assert retrieved is not None
|
|
assert retrieved.edge_type == EdgeType.CALLS
|
|
|
|
def test_remove_node(self):
|
|
g = CodeGraph()
|
|
n = Node(id="x", node_type=NodeType.MODULE, label="x")
|
|
g.add_node(n)
|
|
assert g.has_node("x")
|
|
g.remove_node("x")
|
|
assert not g.has_node("x")
|
|
|
|
def test_successors(self):
|
|
g = CodeGraph()
|
|
g.add_node(Node(id="a", node_type=NodeType.FUNCTION, label="a"))
|
|
g.add_node(Node(id="b", node_type=NodeType.SINK, label="b"))
|
|
g.add_edge(Edge("a", "b", EdgeType.CALLS))
|
|
assert g.successors("a") == ["b"]
|
|
|
|
def test_predecessors(self):
|
|
g = CodeGraph()
|
|
g.add_node(Node(id="a", node_type=NodeType.FUNCTION, label="a"))
|
|
g.add_node(Node(id="b", node_type=NodeType.SINK, label="b"))
|
|
g.add_edge(Edge("a", "b", EdgeType.CALLS))
|
|
assert g.predecessors("b") == ["a"]
|
|
|
|
def test_nodes_by_type(self):
|
|
g = CodeGraph()
|
|
g.add_node(Node(id="m", node_type=NodeType.MODULE, label="m"))
|
|
g.add_node(Node(id="f", node_type=NodeType.FUNCTION, label="f"))
|
|
g.add_node(Node(id="s", node_type=NodeType.SOURCE, label="s"))
|
|
mods = g.nodes_by_type(NodeType.MODULE)
|
|
assert len(mods) == 1
|
|
assert mods[0].id == "m"
|
|
|
|
def test_edges_list(self):
|
|
g = CodeGraph()
|
|
g.add_node(Node(id="a", node_type=NodeType.FUNCTION, label="a"))
|
|
g.add_node(Node(id="b", node_type=NodeType.FUNCTION, label="b"))
|
|
g.add_edge(Edge("a", "b", EdgeType.CALLS))
|
|
edges = g.edges()
|
|
assert len(edges) == 1
|
|
assert edges[0][0] == "a"
|
|
assert edges[0][1] == "b"
|
|
assert edges[0][2].edge_type == EdgeType.CALLS
|
|
|
|
def test_nx_graph(self):
|
|
g = CodeGraph()
|
|
g.add_node(Node(id="a", node_type=NodeType.MODULE, label="a"))
|
|
nxg = g.nx_graph()
|
|
assert nxg.has_node("a")
|
|
|
|
|
|
class TestFlow:
|
|
def test_create_flow(self):
|
|
f = Flow(id="ep::login", name="POST /login", entry_node_id="ep::login")
|
|
assert f.id == "ep::login"
|
|
assert f.name == "POST /login"
|
|
assert f.entry_node_id == "ep::login"
|
|
|
|
def test_flow_defaults(self):
|
|
f = Flow(id="ep::x", name="x", entry_node_id="ep::x")
|
|
assert f.node_ids == []
|
|
assert f.source_node_ids == []
|
|
assert f.sink_node_ids == []
|
|
assert f.tainted_node_ids == []
|
|
assert f.metadata == {}
|
|
|
|
def test_flow_with_lists(self):
|
|
f = Flow(
|
|
id="f",
|
|
name="f",
|
|
entry_node_id="ep",
|
|
node_ids=["a", "b", "c"],
|
|
source_node_ids=["a"],
|
|
sink_node_ids=["c"],
|
|
tainted_node_ids=["b"],
|
|
)
|
|
assert len(f.node_ids) == 3
|
|
assert f.source_node_ids == ["a"]
|
|
assert f.sink_node_ids == ["c"]
|
|
assert f.tainted_node_ids == ["b"]
|