Repository: apache/airflow
File: airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py (Python)

Code before:
        ],
    )
    @pytest.mark.usefixtures("make_primary_connected_component", "make_secondary_connected_component")
    def test_with_node_id_filter(self, test_client, node_id, expected_response_fixture, request):
        expected_response = request.getfixturevalue(expected_response_fixture)
        with assert_queries_count(6):
            response = test_client.get("/dependencies", params={"node_id": node_id})
        assert response.status_code == 200

        assert response.json() == expected_response

    def test_with_node_id_filter_with_asset(
        self,
        test_client,
        asset1_id,
        asset2_id,
        expected_primary_component_response,
        expected_secondary_component_response,
    ):
        for asset_id, expected_response in (
            (asset1_id, expected_primary_component_response),
            (asset2_id, expected_secondary_component_response),
        ):
            with assert_queries_count(6):
                response = test_client.get("/dependencies", params={"node_id": f"asset:{asset_id}"})
            assert response.status_code == 200

            assert response.json() == expected_response

    @pytest.mark.usefixtures("make_primary_connected_component", "make_secondary_connected_component")
    def test_with_node_id_filter_not_found(self, test_client):
        response = test_client.get("/dependencies", params={"node_id": "missing_node_id"})
        assert response.status_code == 404

        assert response.json() == {
            "detail": "Unique connected component not found, got [] for connected components of node missing_node_id, expected only 1 connected component.",
        }

    @pytest.mark.parametrize(
        ("dependency_type", "has_dag", "has_task"),
        [
            (None, True, False),  # default is scheduling
            ("scheduling", True, False),
            ("data", False, True),
        ],
    )
    def test_dependency_type_filter(self, test_client, asset1_id, dependency_type, has_dag, has_task):
        params = {"node_id": f"asset:{asset1_id}"}
        if dependency_type is not None:
            params["dependency_type"] = dependency_type


Review comment (security):
We need an extra test, where we mock the `auth_manager.get_authorized_dag_ids` to return only some dag ids and verify that the response doesn't contain unwanted nodes or edges.

Code after:
            ("dag:downstream_secondary", "expected_secondary_component_response"),
            ("dag:upstream_secondary", "expected_secondary_component_response"),
        ],
    )
    @pytest.mark.usefixtures("make_primary_connected_component", "make_secondary_connected_component")
    def test_with_node_id_filter(self, test_client, node_id, expected_response_fixture, request):
        expected_response = request.getfixturevalue(expected_response_fixture)
        with assert_queries_count(6):
            response = test_client.get("/dependencies", params={"node_id": node_id})
        assert response.status_code == 200

        assert response.json() == expected_response

    def test_with_node_id_filter_with_asset(
        self,
        test_client,
        asset1_id,
        asset2_id,
        expected_primary_component_response,
        expected_secondary_component_response,
    ):
        for asset_id, expected_response in (
            (asset1_id, expected_primary_component_response),
            (asset2_id, expected_secondary_component_response),
        ):
            with assert_queries_count(6):
                response = test_client.get("/dependencies", params={"node_id": f"asset:{asset_id}"})
            assert response.status_code == 200

            assert response.json() == expected_response

    @pytest.mark.usefixtures("make_primary_connected_component", "make_secondary_connected_component")
    def test_with_node_id_filter_not_found(self, test_client):
        response = test_client.get("/dependencies", params={"node_id": "missing_node_id"})
        assert response.status_code == 404

        assert response.json() == {
            "detail": "Unique connected component not found, got [] for connected components of node missing_node_id, expected only 1 connected component.",
        }

    @pytest.mark.parametrize(
        ("dependency_type", "has_dag", "has_task"),
        [
            (None, True, False),  # default is scheduling
            ("scheduling", True, False),
            ("data", False, True),
        ],
    )
    def test_dependency_type_filter(self, test_client, asset1_id, dependency_type, has_dag, has_task):
        params = {"node_id": f"asset:{asset1_id}"}
        if dependency_type is not None: