DataOps-platform/tests/data_research/test_semantic_query.py

from __future__ import annotations

import pytest


class Repository:
    def __init__(self):
        self.calls = []

    def find_property_paths(self, property_uid, *, depth, limit, after_uid):
        self.calls.append((property_uid, depth, limit, after_uid))
        return [
            {
                "path_uid": "path-1",
                "ontology_uid": "ontology-1",
                "ontology_status": "published",
                "property_uid": property_uid,
                "data_element_uid": "element-1",
                "field_uid": "field-1",
                "evidence_uid": "evidence-1",
                "business_domain_uid": "sales",
                "password": "must-not-leak",
            },
            {
                "path_uid": "path-2",
                "ontology_uid": "ontology-draft",
                "ontology_status": "draft",
                "property_uid": property_uid,
                "business_domain_uid": "sales",
            },
            {
                "path_uid": "path-3",
                "ontology_uid": "ontology-hr",
                "ontology_status": "published",
                "property_uid": property_uid,
                "business_domain_uid": "hr",
            },
        ]


def test_semantic_query_is_bounded_published_permission_filtered_and_redacted():
    from app.core.data_research.ontology.query import SemanticQueryService

    repository = Repository()
    result = SemanticQueryService(repository).trace_property(
        "property-1", allowed_domains={"sales"}, limit=10, after_uid=None
    )
    assert [item["path_uid"] for item in result["items"]] == ["path-1"]
    assert "password" not in repr(result).lower()
    assert repository.calls == [("property-1", 3, 11, None)]

    with pytest.raises(ValueError, match="limit"):
        SemanticQueryService(repository).trace_property(
            "property-1", allowed_domains={"sales"}, limit=51
        )


def test_semantic_query_has_stable_cursor_and_no_arbitrary_cypher_interface():
    from app.core.data_research.ontology.query import SemanticQueryService

    service = SemanticQueryService(Repository())
    result = service.trace_property("property-1", allowed_domains={"*"}, limit=1)
    assert result["next_cursor"] == "path-1"
    assert not hasattr(service, "execute_cypher")