kamidso-back/tests/kamidso/test_data_graph_contract.py

import pytest

from kamidso.contracts.errors import KamidsoError
from kamidso.data_graph.explorer.agent import DataMapperAgent, DataMapperMode
from kamidso.data_graph.explorer.contracts import (
    field_id,
    merge_field_semantics,
    normalize_semantic_label,
    normalize_source_map,
)
from kamidso.data_graph.models import needs_semantic_review
from kamidso.data_graph.tools.graph_search import search_maps


def _field(**kwargs):
    payload = {
        "name": "نام",
        "path": ["نام"],
        "address": "excel://employees.xlsx/employees/نام",
        "data_type": "string",
        "confidence": 0.98,
    }
    payload.update(kwargs)
    return payload


def _payload(fields):
    return {
        "kind": "excel",
        "coverage": "observed_partial",
        "datasets": [{"name": "employees", "fields": fields}],
    }


def _normalize(fields, workspace_id="ws_test"):
    return normalize_source_map(
        _payload(fields),
        resource_id="file_hr",
        resource_name="employees.xlsx",
        resource_type="file",
        workspace_id=workspace_id,
    )


def test_clear_header_gets_inferred_label_and_separate_confidence():
    doc = _normalize(
        [
            _field(
                semantic_label="employee_name",
                meaning="نام و نام خانوادگی کارمند.",
                semantic_confidence=0.96,
                semantic_status="inferred",
            )
        ]
    )
    field = doc["datasets"][0]["fields"][0]
    assert field["semantic_label"] == "employee_name"
    assert field["semantic_status"] == "inferred"
    assert field["confidence"] == 0.98
    assert field["semantic_confidence"] == 0.96


def test_unknown_and_needs_review_statuses():
    unknown = _normalize([_field(name="x17", address="excel://employees.xlsx/employees/x17")])["datasets"][0]["fields"][0]
    assert unknown["semantic_status"] == "unknown"
    assert unknown["semantic_label"] == ""
    review = _normalize(
        [
            _field(
                name="x17",
                address="excel://employees.xlsx/employees/x17",
                semantic_label="customer_reference",
                meaning="احتمالاً شناسه طرف حساب",
                semantic_confidence=0.61,
                semantic_status="needs_review",
            )
        ]
    )["datasets"][0]["fields"][0]
    assert review["semantic_status"] == "needs_review"


def test_review_queue_is_only_weak_guesses():
    unknown = _field(name="x17", address="excel://employees.xlsx/employees/x17")
    assert needs_semantic_review(_normalize([unknown])["datasets"][0]["fields"][0]) is False
    weak = _normalize(
        [
            _field(
                semantic_label="customer_reference",
                meaning="احتمالاً شناسه طرف حساب",
                semantic_confidence=0.61,
                semantic_status="needs_review",
            )
        ]
    )["datasets"][0]["fields"][0]
    assert needs_semantic_review(weak) is True
    strong = _normalize(
        [
            _field(
                semantic_label="employee_name",
                meaning="نام کارمند",
                semantic_confidence=0.96,
                semantic_status="inferred",
            )
        ]
    )["datasets"][0]["fields"][0]
    assert needs_semantic_review(strong) is False
    low_inferred = _normalize(
        [
            _field(
                semantic_label="code",
                meaning="احتمالاً کد داخلی",
                semantic_confidence=0.4,
                semantic_status="inferred",
            )
        ]
    )["datasets"][0]["fields"][0]
    assert needs_semantic_review(low_inferred) is True


def test_llm_confirmed_is_coerced_to_inferred():
    field = _normalize(
        [
            _field(
                semantic_label="job_title",
                meaning="عنوان سمت",
                semantic_status="confirmed",
                semantic_confidence=0.9,
            )
        ]
    )["datasets"][0]["fields"][0]
    assert field["semantic_status"] == "inferred"


def test_invalid_label_does_not_fail_the_map():
    field = _normalize(
        [
            _field(
                semantic_label="!!!",
                meaning="معنی مبهم",
                semantic_status="inferred",
            )
        ]
    )["datasets"][0]["fields"][0]
    assert field["semantic_label"] == ""
    assert field["semantic_status"] == "needs_review"
    assert field["meaning"] == "معنی مبهم"


def test_field_id_is_stable_for_the_same_address():
    first = _normalize([_field()])["datasets"][0]["fields"][0]["id"]
    second = _normalize([_field()])["datasets"][0]["fields"][0]["id"]
    assert first == second
    assert first == field_id("ws_test", "file_hr", "excel://employees.xlsx/employees/نام")


def test_rediscovery_keeps_confirmed_and_empty_payload_does_not_wipe():
    confirmed = _normalize(
        [
            _field(
                semantic_label="employee_name",
                meaning="نام کارمند",
                semantic_status="inferred",
            )
        ]
    )
    confirmed["datasets"][0]["fields"][0]["semantic_status"] = "confirmed"
    confirmed["datasets"][0]["fields"][0]["semantic_source"] = "user"
    incoming = _normalize(
        [
            _field(
                semantic_label="full_name",
                meaning="نام کامل",
                semantic_status="inferred",
                semantic_confidence=0.4,
            )
        ]
    )
    merged = merge_field_semantics(confirmed, incoming)
    field = merged["datasets"][0]["fields"][0]
    assert field["semantic_status"] == "confirmed"
    assert field["semantic_label"] == "employee_name"
    empty = _normalize([_field()])
    kept = merge_field_semantics(confirmed, empty)["datasets"][0]["fields"][0]
    assert kept["semantic_label"] == "employee_name"
    assert kept["semantic_status"] == "confirmed"
    incoming["datasets"][0]["fields"][0]["id"] = "fld_new"
    confirmed["datasets"][0]["fields"][0]["id"] = "fld_old"
    remapped = merge_field_semantics(confirmed, incoming)["datasets"][0]["fields"][0]
    assert remapped["id"] == "fld_old"


def test_search_indexes_semantic_label():
    maps = [
        _normalize(
            [_field(semantic_label="payer_name", meaning="", semantic_status="inferred")]
        )
    ]
    maps[0]["status"] = "ready"
    result = search_maps(maps, [], "payer_name")
    assert result["relevant_fields"]
    assert result["relevant_fields"][0]["address"].endswith("/نام")


def test_semantic_mode_tool_names():
    agent = DataMapperAgent(DataMapperMode.SEMANTIC, tools=[], field_contexts=[])
    names = ["get_field_context", "terminate"]
    assert agent.tool_names() == names
    assert "inspect_resource" not in agent.tool_names()
    assert "inspect_path" not in agent.tool_names()
    assert "sample_path" not in agent.tool_names()


def test_normalize_semantic_label_snake_case():
    assert normalize_semantic_label("Payer Name") == "payer_name"
    assert len(normalize_semantic_label("a" * 80)) == 64


def test_coverage_is_still_required():
    with pytest.raises(KamidsoError):
        normalize_source_map(
            {"datasets": [{"name": "t", "fields": [_field()]}]},
            resource_id="r",
            resource_name="r",
            resource_type="file",
        )