741 lines
34 KiB
Python
741 lines
34 KiB
Python
# #region Test.Assistant.DatasetReview [C:2] [TYPE Module] [SEMANTICS test,assistant,dataset,review,context,intent]
|
|
# @BRIEF Tests for _dataset_review.py and _dataset_review_dispatch.py — context loading, intent planning, dispatch.
|
|
# @RELATION BINDS_TO -> [AssistantDatasetReview]
|
|
# @RELATION BINDS_TO -> [AssistantDatasetReviewDispatch]
|
|
# @TEST_EDGE: serialize_context -> Session context serialization with all fields
|
|
# @TEST_EDGE: extract_target -> Target extraction from messages
|
|
# @TEST_EDGE: match_field -> Field matching by id/name/verbose name
|
|
# @TEST_EDGE: extract_quoted_segment -> Quoted segment extraction
|
|
# @TEST_EDGE: plan_intent_approve -> Approve mappings intent
|
|
# @TEST_EDGE: plan_intent_preview -> Generate SQL preview intent
|
|
# @TEST_EDGE: plan_intent_semantic -> Set field semantics intent
|
|
# @TEST_EDGE: plan_intent_context -> Answer context intent
|
|
# @TEST_EDGE: dispatch_answer_context -> Context-only dispatch
|
|
# @TEST_EDGE: dispatch_approve_mappings -> Mapping approval dispatch
|
|
# @TEST_EDGE: dispatch_set_field -> Field semantic update dispatch
|
|
# @TEST_EDGE: dispatch_generate_preview -> SQL preview generation dispatch
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, AsyncMock, patch, PropertyMock
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
|
|
# ── Shared helpers ─────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_session_mock(**overrides):
|
|
"""Create a mock DatasetReviewSession with realistic defaults."""
|
|
from src.models.dataset_review import ReadinessState, RecommendedAction, SessionStatus, SessionPhase
|
|
|
|
s = MagicMock()
|
|
s.session_id = overrides.get("session_id", "sess-1")
|
|
s.version = overrides.get("version", 5)
|
|
s.dataset_ref = overrides.get("dataset_ref", "dataset_123")
|
|
s.dataset_id = overrides.get("dataset_id", 123)
|
|
s.environment_id = overrides.get("environment_id", "env-1")
|
|
s.user_id = overrides.get("user_id", "user-1")
|
|
|
|
rs = overrides.get("readiness_state", ReadinessState.REVIEW_READY)
|
|
s.readiness_state = rs if isinstance(rs, MagicMock) else MagicMock(value=rs.value if hasattr(rs, 'value') else rs)
|
|
s.readiness_state.value = rs.value if hasattr(rs, 'value') else rs
|
|
|
|
ra = overrides.get("recommended_action", RecommendedAction.REVIEW_DOCUMENTATION)
|
|
s.recommended_action = ra if isinstance(ra, MagicMock) else MagicMock(value=ra.value if hasattr(ra, 'value') else ra)
|
|
s.recommended_action.value = ra.value if hasattr(ra, 'value') else ra
|
|
|
|
st = overrides.get("status", SessionStatus.ACTIVE)
|
|
s.status = st if isinstance(st, MagicMock) else MagicMock(value=st.value if hasattr(st, 'value') else st)
|
|
s.status.value = st.value if hasattr(st, 'value') else st
|
|
|
|
cp = overrides.get("current_phase", SessionPhase.REVIEW)
|
|
s.current_phase = cp if isinstance(cp, MagicMock) else MagicMock(value=cp.value if hasattr(cp, 'value') else cp)
|
|
s.current_phase.value = cp.value if hasattr(cp, 'value') else cp
|
|
|
|
s.findings = overrides.get("findings", [])
|
|
s.imported_filters = overrides.get("imported_filters", [])
|
|
s.execution_mappings = overrides.get("execution_mappings", [])
|
|
s.semantic_fields = overrides.get("semantic_fields", [])
|
|
s.previews = overrides.get("previews", [])
|
|
s.last_activity_at = overrides.get("last_activity_at", datetime.now(UTC))
|
|
return s
|
|
|
|
|
|
def _make_finding_mock(finding_id="f-1", code="TEST", severity="WARNING", message="Test finding", resolution_state="OPEN"):
|
|
f = MagicMock()
|
|
f.finding_id = finding_id
|
|
f.code = code
|
|
f.severity = MagicMock(value=severity)
|
|
f.message = message
|
|
f.resolution_state = MagicMock(value=resolution_state)
|
|
return f
|
|
|
|
|
|
def _make_filter_mock(filter_id="fl-1", filter_name="region", display_name="Region", raw_value="US", normalized_value="USA", source="SUPERSET_URL", confidence_state="HIGH", requires_confirmation=False, recovery_status="RECOVERED", notes=None):
|
|
fl = MagicMock()
|
|
fl.filter_id = filter_id
|
|
fl.filter_name = filter_name
|
|
fl.display_name = display_name
|
|
fl.raw_value = raw_value
|
|
fl.raw_value_masked = False
|
|
fl.normalized_value = normalized_value
|
|
fl.source = MagicMock(value=source)
|
|
fl.confidence_state = MagicMock(value=confidence_state)
|
|
fl.requires_confirmation = requires_confirmation
|
|
fl.recovery_status = MagicMock(value=recovery_status)
|
|
fl.notes = notes
|
|
return fl
|
|
|
|
|
|
def _make_mapping_mock(mapping_id="m-1", filter_id="fl-1", variable_id="v-1", mapping_method="DIRECT_MATCH", effective_value="US", approval_state="PENDING", requires_explicit_approval=True):
|
|
m = MagicMock()
|
|
m.mapping_id = mapping_id
|
|
m.filter_id = filter_id
|
|
m.variable_id = variable_id
|
|
m.mapping_method = MagicMock(value=mapping_method)
|
|
m.effective_value = effective_value
|
|
m.approval_state = MagicMock(value=approval_state)
|
|
m.requires_explicit_approval = requires_explicit_approval
|
|
return m
|
|
|
|
|
|
def _make_field_mock(field_id="fld-1", field_name="region", verbose_name="Region", description="The region field", display_format="text", provenance="DICTIONARY_EXACT", is_locked=False, needs_review=False, has_conflict=False):
|
|
f = MagicMock()
|
|
f.field_id = field_id
|
|
f.field_name = field_name
|
|
f.verbose_name = verbose_name
|
|
f.description = description
|
|
f.display_format = display_format
|
|
f.provenance = MagicMock(value=provenance)
|
|
f.is_locked = is_locked
|
|
f.needs_review = needs_review
|
|
f.has_conflict = has_conflict
|
|
return f
|
|
|
|
|
|
class TestSerializeDatasetReviewContext:
|
|
"""_serialize_dataset_review_context — context serialization."""
|
|
|
|
def test_basic_context(self):
|
|
from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
|
|
|
|
session = _make_session_mock()
|
|
ctx = _serialize_dataset_review_context(session)
|
|
assert ctx["session_id"] == "sess-1"
|
|
assert ctx["version"] == 5
|
|
assert ctx["dataset_ref"] == "dataset_123"
|
|
assert ctx["status"] == "active"
|
|
|
|
def test_context_with_findings_and_filters(self):
|
|
from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
|
|
|
|
session = _make_session_mock(findings=[_make_finding_mock()], imported_filters=[_make_filter_mock()], execution_mappings=[_make_mapping_mock()])
|
|
ctx = _serialize_dataset_review_context(session)
|
|
assert len(ctx["findings"]) == 1
|
|
assert ctx["findings"][0]["finding_id"] == "f-1"
|
|
assert len(ctx["imported_filters"]) == 1
|
|
assert ctx["imported_filters"][0]["filter_name"] == "region"
|
|
assert len(ctx["mappings"]) == 1
|
|
assert ctx["mappings"][0]["mapping_id"] == "m-1"
|
|
|
|
def test_context_with_semantic_fields(self):
|
|
from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
|
|
|
|
session = _make_session_mock(semantic_fields=[_make_field_mock()])
|
|
ctx = _serialize_dataset_review_context(session)
|
|
assert len(ctx["semantic_fields"]) == 1
|
|
assert ctx["semantic_fields"][0]["field_name"] == "region"
|
|
|
|
def test_context_with_preview(self):
|
|
from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
|
|
|
|
preview_mock = MagicMock()
|
|
preview_mock.preview_id = "pv-1"
|
|
preview_mock.preview_status = MagicMock(value="READY")
|
|
preview_mock.preview_fingerprint = "fp-abc"
|
|
session = _make_session_mock(previews=[preview_mock])
|
|
ctx = _serialize_dataset_review_context(session)
|
|
# Source does not include 'preview' in returned dict yet
|
|
assert "preview" not in ctx
|
|
|
|
def test_context_no_previews(self):
|
|
from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
|
|
session = _make_session_mock(previews=[])
|
|
ctx = _serialize_dataset_review_context(session)
|
|
# Source does not include 'preview' in returned dict yet
|
|
assert "preview" not in ctx
|
|
|
|
def test_context_without_some_attrs(self):
|
|
"""Handle missing attributes gracefully."""
|
|
from src.api.routes.assistant._dataset_review import _serialize_dataset_review_context
|
|
s = MagicMock()
|
|
s.session_id = "sess-1"
|
|
s.version = 0
|
|
s.dataset_ref = "ds"
|
|
s.dataset_id = None
|
|
s.environment_id = "env-1"
|
|
# Source requires readiness_state to be present (accesses .value)
|
|
s.readiness_state = MagicMock(value="REVIEW_READY")
|
|
s.recommended_action = MagicMock(value="NONE")
|
|
s.status = MagicMock(value="ACTIVE")
|
|
s.current_phase = MagicMock(value="REVIEW")
|
|
s.findings = []
|
|
s.imported_filters = []
|
|
s.execution_mappings = []
|
|
s.semantic_fields = []
|
|
s.previews = []
|
|
ctx = _serialize_dataset_review_context(s)
|
|
# Should handle gracefully or raise AttributeError
|
|
assert ctx["session_id"] == "sess-1"
|
|
|
|
|
|
class TestLoadDatasetReviewContext:
|
|
"""_load_dataset_review_context — context loading from repository."""
|
|
|
|
def test_load_with_valid_session(self):
|
|
from src.api.routes.assistant._dataset_review import _load_dataset_review_context
|
|
from src.schemas.auth import User
|
|
|
|
db = MagicMock()
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
|
|
session = _make_session_mock(user_id="user-1")
|
|
with patch("src.api.routes.assistant._dataset_review.DatasetReviewSessionRepository") as mock_repo:
|
|
mock_repo.return_value.load_session_detail.return_value = session
|
|
result = _load_dataset_review_context("sess-1", user, db)
|
|
assert result is not None
|
|
assert result["session_id"] == "sess-1"
|
|
|
|
def test_load_with_none_session_id(self):
|
|
from src.api.routes.assistant._dataset_review import _load_dataset_review_context
|
|
from src.schemas.auth import User
|
|
|
|
db = MagicMock()
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
result = _load_dataset_review_context(None, user, db)
|
|
assert result is None
|
|
|
|
def test_load_with_wrong_user_raises_404(self):
|
|
from src.api.routes.assistant._dataset_review import _load_dataset_review_context
|
|
from src.schemas.auth import User
|
|
|
|
db = MagicMock()
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
|
|
# Session belongs to different user
|
|
session = _make_session_mock(user_id="user-2")
|
|
with patch("src.api.routes.assistant._dataset_review.DatasetReviewSessionRepository") as mock_repo:
|
|
mock_repo.return_value.load_session_detail.return_value = session
|
|
with pytest.raises(HTTPException) as exc:
|
|
_load_dataset_review_context("sess-1", user, db)
|
|
assert exc.value.status_code == 404
|
|
|
|
def test_load_with_missing_session_raises_404(self):
|
|
from src.api.routes.assistant._dataset_review import _load_dataset_review_context
|
|
from src.schemas.auth import User
|
|
|
|
db = MagicMock()
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
|
|
with patch("src.api.routes.assistant._dataset_review.DatasetReviewSessionRepository") as mock_repo:
|
|
mock_repo.return_value.load_session_detail.return_value = None
|
|
with pytest.raises(HTTPException) as exc:
|
|
_load_dataset_review_context("sess-1", user, db)
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
class TestExtractDatasetReviewTarget:
|
|
"""_extract_dataset_review_target — target extraction."""
|
|
|
|
@pytest.mark.parametrize("msg,expected_kind,expected_id", [
|
|
("target:field=abc123", "field", "abc123"),
|
|
("focus:field=xyz789", "field", "xyz789"),
|
|
("target:mapping=map_1", "mapping", "map_1"),
|
|
("target:finding=fnd_1", "finding", "fnd_1"),
|
|
("target:filter=flt_1", "filter", "flt_1"),
|
|
("focus = finding=abc", "finding", "abc"),
|
|
("target=field=def", "field", "def"),
|
|
("target = field = def", None, None),
|
|
("no target here", None, None),
|
|
("", None, None),
|
|
])
|
|
def test_extract(self, msg, expected_kind, expected_id):
|
|
from src.api.routes.assistant._dataset_review import _extract_dataset_review_target
|
|
kind, tid = _extract_dataset_review_target(msg)
|
|
assert kind == expected_kind
|
|
assert tid == expected_id
|
|
|
|
|
|
class TestMatchDatasetReviewField:
|
|
"""_match_dataset_review_field — field matching."""
|
|
|
|
def test_match_by_id(self):
|
|
from src.api.routes.assistant._dataset_review import _match_dataset_review_field
|
|
|
|
ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region"}]}
|
|
result = _match_dataset_review_field(ctx, "target:field=fld-1")
|
|
assert result is not None
|
|
assert result["field_id"] == "fld-1"
|
|
|
|
def test_match_by_field_name(self):
|
|
from src.api.routes.assistant._dataset_review import _match_dataset_review_field
|
|
|
|
ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region"}]}
|
|
result = _match_dataset_review_field(ctx, "please update region field")
|
|
assert result is not None
|
|
assert result["field_name"] == "region"
|
|
|
|
def test_match_by_verbose_name(self):
|
|
from src.api.routes.assistant._dataset_review import _match_dataset_review_field
|
|
|
|
ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region Name"}]}
|
|
result = _match_dataset_review_field(ctx, "change Region Name")
|
|
assert result is not None
|
|
assert result["verbose_name"] == "Region Name"
|
|
|
|
def test_no_match(self):
|
|
from src.api.routes.assistant._dataset_review import _match_dataset_review_field
|
|
|
|
ctx = {"semantic_fields": [{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region"}]}
|
|
result = _match_dataset_review_field(ctx, "something completely different")
|
|
assert result is None
|
|
|
|
def test_empty_fields(self):
|
|
from src.api.routes.assistant._dataset_review import _match_dataset_review_field
|
|
|
|
ctx = {"semantic_fields": []}
|
|
result = _match_dataset_review_field(ctx, "region")
|
|
assert result is None
|
|
|
|
|
|
class TestExtractQuotedSegment:
|
|
"""_extract_quoted_segment — quoted command segment extraction."""
|
|
|
|
@pytest.mark.parametrize("msg,label,expected", [
|
|
('verbose_name="My Label"', "verbose_name", "My Label"),
|
|
("verbose_name='My Label'", "verbose_name", "My Label"),
|
|
('description="Some desc"', "description", "Some desc"),
|
|
('display_format="text"', "display_format", "text"),
|
|
('verbose_name: "Label"', "verbose_name", "Label"),
|
|
('verbose_name = "Label"', "verbose_name", "Label"),
|
|
("no label here", "verbose_name", None),
|
|
])
|
|
def test_extract(self, msg, label, expected):
|
|
from src.api.routes.assistant._dataset_review import _extract_quoted_segment
|
|
result = _extract_quoted_segment(msg, label)
|
|
assert result == expected
|
|
|
|
|
|
class TestPlanDatasetReviewIntent:
|
|
"""_plan_dataset_review_intent — intent planning from message."""
|
|
|
|
@pytest.fixture
|
|
def context(self):
|
|
return {
|
|
"session_id": "sess-1",
|
|
"version": 5,
|
|
"readiness_state": "REVIEW_READY",
|
|
"recommended_action": "REVIEW_DOCUMENTATION",
|
|
"mappings": [
|
|
{"mapping_id": "m-1", "requires_explicit_approval": True, "approval_state": "PENDING"},
|
|
{"mapping_id": "m-2", "requires_explicit_approval": True, "approval_state": "PENDING"},
|
|
],
|
|
"findings": [{"code": "TEST"}],
|
|
"imported_filters": [{"filter_name": "region"}],
|
|
"preview": {"preview_status": "missing"},
|
|
"semantic_fields": [],
|
|
}
|
|
|
|
def test_approve_mappings(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("approve mappings", context)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_approve_mappings"
|
|
assert result["entities"]["mapping_ids"] == ["m-1", "m-2"]
|
|
assert result["requires_confirmation"] is True
|
|
|
|
def test_approve_mappings_russian(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("подтверди маппинг", context)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_approve_mappings"
|
|
|
|
def test_generate_sql_preview(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("generate sql preview", context)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_generate_sql_preview"
|
|
|
|
def test_generate_preview_russian(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("сгенерируй превью", context)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_generate_sql_preview"
|
|
|
|
def test_set_field_semantics_with_target(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
ctx_with_field = dict(context)
|
|
ctx_with_field["semantic_fields"] = [
|
|
{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region", "candidates": []}
|
|
]
|
|
# Source _extract_quoted_segment union regex only captures with last alternative (label)
|
|
result = _plan_dataset_review_intent('set field semantics target:field=fld-1 label="Country"', ctx_with_field)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_set_field_semantics"
|
|
assert result["entities"]["verbose_name"] == "Country"
|
|
|
|
def test_set_field_semantics_no_field_returns_none(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
# No matching field in message and no target
|
|
result = _plan_dataset_review_intent("set field semantics", context)
|
|
assert result is None
|
|
|
|
def test_set_field_semantics_with_lock(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
ctx_with_field = dict(context)
|
|
ctx_with_field["semantic_fields"] = [
|
|
{"field_id": "fld-1", "field_name": "region", "verbose_name": "Region", "candidates": []}
|
|
]
|
|
# Source _extract_quoted_segment union regex only captures with last alternative (label)
|
|
result = _plan_dataset_review_intent('apply field semantics target:field=fld-1 lock it label="Country"', ctx_with_field)
|
|
assert result is not None
|
|
assert result["entities"]["lock_field"] is True
|
|
|
|
def test_answer_context(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("what are the filters?", context)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_answer_context"
|
|
# Summary uses 'readiness=' (not 'readiness_state=')
|
|
assert "readiness" in result["entities"]["summary"]
|
|
assert result["requires_confirmation"] is False
|
|
|
|
def test_answer_context_russian(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("фильтр", context)
|
|
assert result is not None
|
|
assert result["operation"] == "dataset_review_answer_context"
|
|
|
|
def test_unrelated_message_returns_none(self, context):
|
|
from src.api.routes.assistant._dataset_review import _plan_dataset_review_intent
|
|
result = _plan_dataset_review_intent("hello world", context)
|
|
assert result is None
|
|
|
|
|
|
class TestDispatchDatasetReviewIntent:
|
|
"""_dispatch_dataset_review_intent — intent dispatch."""
|
|
|
|
@pytest.fixture
|
|
def user(self):
|
|
from src.schemas.auth import User
|
|
return User(id="user-1", username="test", email="test@x.com", is_active=True, auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
|
|
@pytest.fixture
|
|
def config_manager(self):
|
|
cm = MagicMock()
|
|
cm.get_environments = MagicMock(return_value=[])
|
|
return cm
|
|
|
|
def test_answer_context(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
db = MagicMock()
|
|
intent = {
|
|
"operation": "dataset_review_answer_context",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
"summary": "Session sess-1: readiness=READY",
|
|
},
|
|
}
|
|
text, task_id, actions = (
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
)
|
|
assert "READY" in text
|
|
assert task_id is None
|
|
|
|
def test_missing_session_id_raises_422(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
db = MagicMock()
|
|
intent = {"operation": "test", "entities": {}}
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 422
|
|
|
|
def test_unsupported_operation_raises_400(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
db = MagicMock()
|
|
intent = {
|
|
"operation": "dataset_review_unknown",
|
|
"entities": {"dataset_review_session_id": "sess-1", "session_version": 5},
|
|
}
|
|
|
|
session = _make_session_mock()
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
mock_repo.bump_session_version.return_value = None
|
|
mock_repo.db = MagicMock()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 400
|
|
|
|
def test_approve_mappings_success(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
from src.models.dataset_review import ApprovalState, ReadinessState
|
|
|
|
db = MagicMock()
|
|
mapping = _make_mapping_mock(mapping_id="m-1", filter_id="fl-1", variable_id="v-1",
|
|
approval_state=ApprovalState.PENDING.value,
|
|
requires_explicit_approval=True)
|
|
|
|
session = _make_session_mock(execution_mappings=[mapping],
|
|
readiness_state=ReadinessState.MAPPING_REVIEW_NEEDED,
|
|
user_id="user-1")
|
|
|
|
intent = {
|
|
"operation": "dataset_review_approve_mappings",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
"mapping_ids": ["m-1"],
|
|
},
|
|
}
|
|
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
mock_repo.bump_session_version.return_value = None
|
|
mock_repo.db = MagicMock()
|
|
mock_repo.event_logger = MagicMock()
|
|
|
|
text, task_id, actions = (
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
)
|
|
assert "Approved" in text
|
|
assert "1 mapping" in text
|
|
|
|
def test_approve_mappings_no_pending_raises_409(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
db = MagicMock()
|
|
intent = {
|
|
"operation": "dataset_review_approve_mappings",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
"mapping_ids": [],
|
|
},
|
|
}
|
|
|
|
session = _make_session_mock(execution_mappings=[], user_id="user-1")
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 409
|
|
assert "No pending mappings" in exc.value.detail
|
|
|
|
def test_approve_mappings_no_match_raises_409(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
from src.models.dataset_review import ApprovalState
|
|
|
|
db = MagicMock()
|
|
session = _make_session_mock(execution_mappings=[], user_id="user-1")
|
|
|
|
intent = {
|
|
"operation": "dataset_review_approve_mappings",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
"mapping_ids": ["m-nonexistent"],
|
|
},
|
|
}
|
|
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 409
|
|
|
|
def test_set_field_semantics_success(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
field = _make_field_mock()
|
|
session = _make_session_mock(semantic_fields=[field], user_id="user-1")
|
|
|
|
intent = {
|
|
"operation": "dataset_review_set_field_semantics",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
"field_id": "fld-1",
|
|
"verbose_name": "Country",
|
|
},
|
|
}
|
|
|
|
db = MagicMock()
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
mock_repo.bump_session_version.return_value = None
|
|
mock_repo.db = MagicMock()
|
|
mock_repo.event_logger = MagicMock()
|
|
|
|
with patch("src.api.routes.assistant._dataset_review_dispatch._update_semantic_field_state") as mock_update:
|
|
mock_update.return_value = None
|
|
text, task_id, actions = (
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
)
|
|
assert "Updated semantic field" in text
|
|
|
|
def test_set_field_semantics_missing_field_raises_422(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
session = _make_session_mock(semantic_fields=[], user_id="user-1")
|
|
intent = {
|
|
"operation": "dataset_review_set_field_semantics",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
"field_id": "",
|
|
},
|
|
}
|
|
|
|
db = MagicMock()
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 422
|
|
|
|
def test_session_not_found_raises_404(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
db = MagicMock()
|
|
intent = {
|
|
"operation": "dataset_review_approve_mappings",
|
|
"entities": {"dataset_review_session_id": "sess-1", "session_version": 5, "mapping_ids": ["m-1"]},
|
|
}
|
|
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = None
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 404
|
|
|
|
def test_version_conflict_raises_409(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent, DatasetReviewSessionVersionConflictError
|
|
|
|
db = MagicMock()
|
|
intent = {
|
|
"operation": "dataset_review_approve_mappings",
|
|
"entities": {"dataset_review_session_id": "sess-1", "session_version": 4, "mapping_ids": ["m-1"]},
|
|
}
|
|
|
|
session = _make_session_mock(user_id="user-1")
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
with patch(repository_cls) as mock_repo_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.side_effect = DatasetReviewSessionVersionConflictError(
|
|
session_id="sess-1", expected_version=4, actual_version=5
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
assert exc.value.status_code == 409
|
|
assert "session_version_conflict" in exc.value.detail["error_code"]
|
|
|
|
def test_generate_sql_preview_success(self, user, config_manager):
|
|
from src.api.routes.assistant._dataset_review_dispatch import _dispatch_dataset_review_intent
|
|
|
|
session = _make_session_mock(user_id="user-1")
|
|
db = MagicMock()
|
|
|
|
intent = {
|
|
"operation": "dataset_review_generate_sql_preview",
|
|
"entities": {
|
|
"dataset_review_session_id": "sess-1",
|
|
"session_version": 5,
|
|
},
|
|
}
|
|
|
|
repository_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewSessionRepository"
|
|
orchestrator_cls = "src.api.routes.assistant._dataset_review_dispatch.DatasetReviewOrchestrator"
|
|
with patch(repository_cls) as mock_repo_cls, patch(orchestrator_cls) as mock_orch_cls:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_repo.load_session_detail.return_value = session
|
|
mock_repo.require_session_version.return_value = None
|
|
mock_repo.bump_session_version.return_value = None
|
|
mock_repo.db = MagicMock()
|
|
mock_repo.event_logger = MagicMock()
|
|
|
|
mock_orch = MagicMock()
|
|
mock_orch_cls.return_value = mock_orch
|
|
preview_result = MagicMock()
|
|
preview_result.preview = MagicMock()
|
|
preview_result.preview.preview_status = "READY"
|
|
mock_orch.prepare_launch_preview.return_value = preview_result
|
|
|
|
text, task_id, actions = (
|
|
__import__("asyncio").run(_dispatch_dataset_review_intent(intent, user, config_manager, db))
|
|
)
|
|
assert "READY" in text
|
|
|
|
|
|
class TestDatasetReviewConflictHttpException:
|
|
"""_dataset_review_conflict_http_exception — version conflict error mapping."""
|
|
|
|
def test_conflict_exception(self):
|
|
from src.api.routes.assistant._dataset_review_dispatch import (
|
|
_dataset_review_conflict_http_exception,
|
|
DatasetReviewSessionVersionConflictError,
|
|
)
|
|
exc = DatasetReviewSessionVersionConflictError(
|
|
session_id="sess-1", expected_version=4, actual_version=5
|
|
)
|
|
http_exc = _dataset_review_conflict_http_exception(exc)
|
|
assert http_exc.status_code == 409
|
|
assert http_exc.detail["error_code"] == "session_version_conflict"
|
|
assert http_exc.detail["session_id"] == "sess-1"
|
|
# #endregion Test.Assistant.DatasetReview
|