q3alique-codeflow/tests/unit/test_python_extractor.py

"""Tasks 006-008 tests: Python extractor (functions, sources, sinks)."""
import pytest
from pathlib import Path
from codeflow.discovery.file_scanner import FileInfo
from codeflow.analysis.python_extractor import extract, PythonExtractor
from codeflow.models.node import NodeType
from tests.fixtures import FLASK_FIXTURE_DIR


@pytest.fixture
def helpers_fixture():
    return FileInfo(
        path=str(FLASK_FIXTURE_DIR / "helpers.py"),
        language="python",
        size_bytes=0,
        relative_path="flask_app/helpers.py",
    )


@pytest.fixture
def flask_app_fixture():
    return FileInfo(
        path=str(FLASK_FIXTURE_DIR / "app.py"),
        language="python",
        size_bytes=0,
        relative_path="flask_app/app.py",
    )


@pytest.fixture
def auth_fixture():
    return FileInfo(
        path=str(FLASK_FIXTURE_DIR / "auth.py"),
        language="python",
        size_bytes=0,
        relative_path="flask_app/auth.py",
    )


class TestPythonExtractor:
    def test_module_node_created(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        mods = [n for n in result.nodes if n.node_type == NodeType.MODULE]
        assert len(mods) == 1
        assert mods[0].label == "app.py"

    def test_entry_points_count(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        eps = [n for n in result.nodes if n.node_type == NodeType.ENTRY_POINT]
        assert len(eps) >= 3

    def test_login_entry_point(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        login = None
        for n in result.nodes:
            if n.node_type == NodeType.ENTRY_POINT and "/login" in n.label:
                login = n
                break
        assert login is not None
        assert login.metadata.get("http_method") == "POST"
        assert login.metadata.get("route_path") == "/login"

    def test_username_source_node(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        sources = [n for n in result.nodes if n.node_type == NodeType.SOURCE]
        usernames = [s for s in sources if "username" in s.label]
        assert len(usernames) >= 1

    def test_username_input_type(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        sources = [n for n in result.nodes if n.node_type == NodeType.SOURCE]
        for s in sources:
            if "username" in s.label:
                assert s.metadata.get("input_type") == "form_field"
                return
        pytest.fail("No username source found")

    def test_database_sink(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        sinks = [n for n in result.nodes if n.node_type == NodeType.SINK]
        db_sinks = [s for s in sinks if s.metadata.get("sink_type") == "database"]
        assert len(db_sinks) >= 1

    def test_command_execution_sink(self, helpers_fixture):
        result = extract(helpers_fixture)
        sinks = [n for n in result.nodes if n.node_type == NodeType.SINK]
        cmd_sinks = [s for s in sinks if s.metadata.get("sink_type") == "command_execution"]
        assert len(cmd_sinks) >= 1

    def test_function_node_exists(self, auth_fixture):
        result = extract(auth_fixture)
        funcs = [n for n in result.nodes if n.node_type == NodeType.FUNCTION]
        names = [f.label for f in funcs]
        assert "authenticate" in names

    def test_calls_edge(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        from codeflow.models.edge import EdgeType
        calls = [e for e in result.edges if e.edge_type == EdgeType.CALLS]
        assert len(calls) >= 1

    def test_data_flow_edge(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        from codeflow.models.edge import EdgeType
        data_flows = [e for e in result.edges if e.edge_type == EdgeType.DATA_FLOW]
        assert len(data_flows) >= 1

    def test_flask_import_detected(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        mods = [n for n in result.nodes if n.node_type == NodeType.MODULE]
        assert len(mods) == 1
        imports = mods[0].metadata.get("imports", [])
        has_flask = any("flask" in imp.lower() for imp in imports)
        assert has_flask

    def test_no_functions_file(self, tmp_path):
        f = tmp_path / "empty_mod.py"
        f.write_text("x = 1\ny = 2")
        fi = FileInfo(path=str(f), language="python", size_bytes=0, relative_path="empty_mod.py")
        result = extract(fi)
        mods = [n for n in result.nodes if n.node_type == NodeType.MODULE]
        assert len(mods) == 1
        funcs = [n for n in result.nodes if n.node_type == NodeType.FUNCTION]
        assert len(funcs) == 0

    def test_syntax_error_file(self, tmp_path):
        f = tmp_path / "broken.py"
        f.write_text("def broken(")
        fi = FileInfo(path=str(f), language="python", size_bytes=0, relative_path="broken.py")
        result = extract(fi)
        assert len(result.errors) >= 0

    def test_unique_node_ids(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        ids = [n.id for n in result.nodes]
        assert len(ids) == len(set(ids))

    def test_env_var_node(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        env_vars = [n for n in result.nodes if n.node_type == NodeType.ENV_VAR]
        assert len(env_vars) >= 1

    def test_run_code_sink(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        sinks = [n for n in result.nodes if n.node_type == NodeType.SINK]
        eval_sinks = [s for s in sinks if s.metadata.get("sink_type") == "code_execution"]
        assert len(eval_sinks) >= 1

    def test_file_write_sink(self, flask_app_fixture):
        result = extract(flask_app_fixture)
        sinks = [n for n in result.nodes if n.node_type == NodeType.SINK]
        fw_sinks = [s for s in sinks if s.metadata.get("sink_type") == "file_write"]
        assert len(fw_sinks) >= 1