511 lines
20 KiB
Python
511 lines
20 KiB
Python
# #region Test.Schemas.DatasetReview [C:2] [TYPE Module] [SEMANTICS test,schema,dataset,review]
|
|
# @BRIEF Tests for schemas/dataset_review.py and dataset_review_pkg/* — DTOs and composites.
|
|
# @RELATION BINDS_TO -> [DatasetReviewSchemas]
|
|
# @TEST_EDGE: missing_field -> optional fields default correctly
|
|
# @TEST_EDGE: invalid_type -> Pydantic validation rejects bad types
|
|
# @TEST_EDGE: serialization -> round-trip via model_dump/model_validate
|
|
|
|
from datetime import datetime, timezone
|
|
import pytest
|
|
|
|
# ── DTOs ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSessionCollaboratorDto:
|
|
"""SessionCollaboratorDto — from dataset_review_pkg._dtos."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SessionCollaboratorDto
|
|
from src.models.dataset_review import SessionCollaboratorRole
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SessionCollaboratorDto(
|
|
user_id="u-1",
|
|
role=SessionCollaboratorRole.VIEWER,
|
|
added_at=ts,
|
|
)
|
|
assert dto.user_id == "u-1"
|
|
assert dto.role == SessionCollaboratorRole.VIEWER
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SessionCollaboratorDto
|
|
from src.models.dataset_review import SessionCollaboratorRole
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SessionCollaboratorDto(user_id="u-1", role=SessionCollaboratorRole.REVIEWER, added_at=ts)
|
|
data = dto.model_dump()
|
|
restored = SessionCollaboratorDto.model_validate(data)
|
|
assert restored.user_id == "u-1"
|
|
assert restored.role == "reviewer"
|
|
|
|
def test_from_attributes(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SessionCollaboratorDto
|
|
from src.models.dataset_review import SessionCollaboratorRole
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
|
|
class FakeORM:
|
|
user_id = "u-1"
|
|
role = SessionCollaboratorRole.APPROVER
|
|
added_at = ts
|
|
dto = SessionCollaboratorDto.model_validate(FakeORM())
|
|
assert dto.user_id == "u-1"
|
|
|
|
|
|
class TestDatasetProfileDto:
|
|
"""DatasetProfileDto — full profile DTO."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import DatasetProfileDto
|
|
from src.models.dataset_review import BusinessSummarySource, ConfidenceState
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = DatasetProfileDto(
|
|
profile_id="p-1",
|
|
session_id="s-1",
|
|
dataset_name="sales",
|
|
business_summary="Sales data",
|
|
business_summary_source=BusinessSummarySource.IMPORTED,
|
|
is_sqllab_view=False,
|
|
confidence_state=ConfidenceState.CONFIRMED,
|
|
has_blocking_findings=False,
|
|
has_warning_findings=False,
|
|
manual_summary_locked=False,
|
|
created_at=ts,
|
|
updated_at=ts,
|
|
)
|
|
assert dto.profile_id == "p-1"
|
|
assert dto.completeness_score is None
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.schemas.dataset_review_pkg._dtos import DatasetProfileDto
|
|
from src.models.dataset_review import BusinessSummarySource, ConfidenceState
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = DatasetProfileDto(
|
|
profile_id="p-1", session_id="s-1", dataset_name="sales",
|
|
business_summary="Sales data", business_summary_source=BusinessSummarySource.INFERRED,
|
|
is_sqllab_view=False, confidence_state=ConfidenceState.MIXED,
|
|
has_blocking_findings=False, has_warning_findings=False,
|
|
manual_summary_locked=False, created_at=ts, updated_at=ts,
|
|
)
|
|
data = dto.model_dump()
|
|
restored = DatasetProfileDto.model_validate(data)
|
|
assert restored.profile_id == "p-1"
|
|
|
|
|
|
class TestValidationFindingDto:
|
|
"""ValidationFindingDto — finding with resolution tracking."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import ValidationFindingDto
|
|
from src.models.dataset_review import FindingArea, FindingSeverity, ResolutionState
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ValidationFindingDto(
|
|
finding_id="f-1",
|
|
session_id="s-1",
|
|
area=FindingArea.DATASET_PROFILE,
|
|
severity=FindingSeverity.BLOCKING,
|
|
code="DQ001",
|
|
title="Null values",
|
|
message="Found null values",
|
|
resolution_state=ResolutionState.OPEN,
|
|
created_at=ts,
|
|
)
|
|
assert dto.finding_id == "f-1"
|
|
assert dto.resolved_at is None
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.schemas.dataset_review_pkg._dtos import ValidationFindingDto
|
|
from src.models.dataset_review import FindingArea, FindingSeverity, ResolutionState
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ValidationFindingDto(
|
|
finding_id="f-1", session_id="s-1", area=FindingArea.SOURCE_INTAKE,
|
|
severity=FindingSeverity.WARNING, code="DQ002", title="Duplicates",
|
|
message="Found duplicates", resolution_state=ResolutionState.RESOLVED,
|
|
resolved_at=ts, created_at=ts,
|
|
)
|
|
data = dto.model_dump()
|
|
restored = ValidationFindingDto.model_validate(data)
|
|
assert restored.code == "DQ002"
|
|
|
|
|
|
class TestSemanticSourceDto:
|
|
"""SemanticSourceDto — source with trust level."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SemanticSourceDto
|
|
from src.models.dataset_review import SemanticSourceStatus, SemanticSourceType, TrustLevel
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SemanticSourceDto(
|
|
source_id="s-1", session_id="ses-1",
|
|
source_type=SemanticSourceType.REFERENCE_DATASET,
|
|
source_ref="ref1", source_version="1.0",
|
|
display_name="Reference Dataset",
|
|
trust_level=TrustLevel.TRUSTED,
|
|
status=SemanticSourceStatus.AVAILABLE,
|
|
created_at=ts,
|
|
)
|
|
assert dto.source_id == "s-1"
|
|
assert dto.schema_overlap_score is None
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SemanticSourceDto
|
|
from src.models.dataset_review import SemanticSourceStatus, SemanticSourceType, TrustLevel
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SemanticSourceDto(
|
|
source_id="s-1", session_id="ses-1",
|
|
source_type=SemanticSourceType.CONNECTED_DICTIONARY,
|
|
source_ref="ref1", source_version="1.0",
|
|
display_name="Dictionary", trust_level=TrustLevel.RECOMMENDED,
|
|
status=SemanticSourceStatus.SELECTED, created_at=ts,
|
|
)
|
|
data = dto.model_dump()
|
|
restored = SemanticSourceDto.model_validate(data)
|
|
assert restored.trust_level == "recommended"
|
|
|
|
|
|
class TestSemanticCandidateDto:
|
|
"""SemanticCandidateDto — candidate with match type."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto
|
|
from src.models.dataset_review import CandidateMatchType, CandidateStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SemanticCandidateDto(
|
|
candidate_id="c-1", field_id="f-1",
|
|
candidate_rank=1, match_type=CandidateMatchType.EXACT,
|
|
confidence_score=0.95, status=CandidateStatus.ACCEPTED,
|
|
created_at=ts,
|
|
)
|
|
assert dto.confidence_score == 0.95
|
|
assert dto.source_id is None
|
|
|
|
def test_serialize_roundtrip(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto
|
|
from src.models.dataset_review import CandidateMatchType, CandidateStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SemanticCandidateDto(
|
|
candidate_id="c-1", field_id="f-1", candidate_rank=1,
|
|
match_type=CandidateMatchType.FUZZY, confidence_score=0.75,
|
|
status=CandidateStatus.PROPOSED, created_at=ts,
|
|
)
|
|
data = dto.model_dump()
|
|
restored = SemanticCandidateDto.model_validate(data)
|
|
assert restored.status == "proposed"
|
|
|
|
|
|
class TestSemanticFieldEntryDto:
|
|
"""SemanticFieldEntryDto — field with nested candidates."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SemanticFieldEntryDto
|
|
from src.models.dataset_review import FieldKind, FieldProvenance
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SemanticFieldEntryDto(
|
|
field_id="f-1", session_id="s-1",
|
|
field_name="amount", field_kind=FieldKind.METRIC,
|
|
provenance=FieldProvenance.FUZZY_INFERRED,
|
|
is_locked=False, has_conflict=False, needs_review=False,
|
|
last_changed_by="admin", created_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.candidates == []
|
|
|
|
def test_with_candidates(self):
|
|
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto, SemanticFieldEntryDto
|
|
from src.models.dataset_review import CandidateMatchType, CandidateStatus, FieldKind, FieldProvenance
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
cand = SemanticCandidateDto(
|
|
candidate_id="c-1", field_id="f-1", candidate_rank=1,
|
|
match_type=CandidateMatchType.EXACT, confidence_score=0.99,
|
|
status=CandidateStatus.ACCEPTED, created_at=ts,
|
|
)
|
|
dto = SemanticFieldEntryDto(
|
|
field_id="f-1", session_id="s-1", field_name="revenue",
|
|
field_kind=FieldKind.METRIC, provenance=FieldProvenance.DICTIONARY_EXACT,
|
|
is_locked=True, has_conflict=False, needs_review=False,
|
|
last_changed_by="admin", candidates=[cand],
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
assert len(dto.candidates) == 1
|
|
assert dto.candidates[0].confidence_score == 0.99
|
|
|
|
|
|
class TestImportedFilterDto:
|
|
"""ImportedFilterDto — filter with raw_value and confidence."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import ImportedFilterDto
|
|
from src.models.dataset_review import FilterConfidenceState, FilterRecoveryStatus, FilterSource
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ImportedFilterDto(
|
|
filter_id="fl-1", session_id="s-1",
|
|
filter_name="region", raw_value="EMEA",
|
|
source=FilterSource.SUPERSET_URL,
|
|
confidence_state=FilterConfidenceState.CONFIRMED,
|
|
requires_confirmation=False,
|
|
recovery_status=FilterRecoveryStatus.RECOVERED,
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.filter_name == "region"
|
|
assert dto.normalized_value is None
|
|
|
|
def test_with_raw_value_masked(self):
|
|
from src.schemas.dataset_review_pkg._dtos import ImportedFilterDto
|
|
from src.models.dataset_review import FilterConfidenceState, FilterRecoveryStatus, FilterSource
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ImportedFilterDto(
|
|
filter_id="fl-1", session_id="s-1",
|
|
filter_name="region", raw_value="***", raw_value_masked=True,
|
|
source=FilterSource.SUPERSET_NATIVE,
|
|
confidence_state=FilterConfidenceState.IMPORTED,
|
|
requires_confirmation=True,
|
|
recovery_status=FilterRecoveryStatus.PARTIAL,
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.raw_value_masked is True
|
|
assert dto.requires_confirmation is True
|
|
|
|
|
|
class TestTemplateVariableDto:
|
|
"""TemplateVariableDto — variable with mapping status."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import TemplateVariableDto
|
|
from src.models.dataset_review import MappingStatus, VariableKind
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = TemplateVariableDto(
|
|
variable_id="v-1", session_id="s-1",
|
|
variable_name="country", expression_source="{{ country }}",
|
|
variable_kind=VariableKind.NATIVE_FILTER,
|
|
is_required=True, mapping_status=MappingStatus.UNMAPPED,
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.variable_name == "country"
|
|
assert dto.default_value is None
|
|
|
|
|
|
class TestExecutionMappingDto:
|
|
"""ExecutionMappingDto — mapping with approval state."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._dtos import ExecutionMappingDto
|
|
from src.models.dataset_review import ApprovalState, MappingMethod
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ExecutionMappingDto(
|
|
mapping_id="m-1", session_id="s-1",
|
|
filter_id="f-1", variable_id="v-1",
|
|
mapping_method=MappingMethod.DIRECT_MATCH,
|
|
raw_input_value="EMEA",
|
|
requires_explicit_approval=False,
|
|
approval_state=ApprovalState.NOT_REQUIRED,
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.mapping_method == "direct_match"
|
|
assert dto.effective_value is None
|
|
|
|
|
|
# ── Composites ──────────────────────────────────────────────
|
|
|
|
|
|
class TestClarificationOptionDto:
|
|
"""ClarificationOptionDto — option for clarification question."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._composites import ClarificationOptionDto
|
|
|
|
dto = ClarificationOptionDto(
|
|
option_id="o-1", question_id="q-1",
|
|
label="Yes", value="yes",
|
|
is_recommended=True, display_order=1,
|
|
)
|
|
assert dto.label == "Yes"
|
|
assert dto.is_recommended is True
|
|
|
|
|
|
class TestClarificationAnswerDto:
|
|
"""ClarificationAnswerDto — answer with feedback."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._composites import ClarificationAnswerDto
|
|
from src.models.dataset_review import AnswerKind
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ClarificationAnswerDto(
|
|
answer_id="a-1", question_id="q-1",
|
|
answer_kind=AnswerKind.SELECTED,
|
|
answered_by_user_id="u-1",
|
|
created_at=ts,
|
|
)
|
|
assert dto.impact_summary is None
|
|
|
|
|
|
class TestClarificationQuestionDto:
|
|
"""ClarificationQuestionDto — question with nested options and answer."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._composites import ClarificationQuestionDto
|
|
from src.models.dataset_review import QuestionState
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ClarificationQuestionDto(
|
|
question_id="q-1", clarification_session_id="cs-1",
|
|
topic_ref="topic1", question_text="What?",
|
|
why_it_matters="Need to know", priority=1,
|
|
state=QuestionState.OPEN, created_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.options == []
|
|
assert dto.answer is None
|
|
|
|
|
|
class TestClarificationSessionDto:
|
|
"""ClarificationSessionDto — session with nested questions."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._composites import ClarificationSessionDto
|
|
from src.models.dataset_review import ClarificationStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = ClarificationSessionDto(
|
|
clarification_session_id="cs-1", session_id="s-1",
|
|
status=ClarificationStatus.ACTIVE,
|
|
resolved_count=0, remaining_count=5,
|
|
started_at=ts, updated_at=ts,
|
|
)
|
|
assert dto.questions == []
|
|
|
|
|
|
class TestCompiledPreviewDto:
|
|
"""CompiledPreviewDto — preview with fingerprint."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._composites import CompiledPreviewDto
|
|
from src.models.dataset_review import PreviewStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = CompiledPreviewDto(
|
|
preview_id="p-1", session_id="s-1",
|
|
preview_status=PreviewStatus.READY,
|
|
preview_fingerprint="abc123", compiled_by="admin",
|
|
compiled_at=ts, created_at=ts,
|
|
)
|
|
assert dto.error_code is None
|
|
|
|
|
|
class TestDatasetRunContextDto:
|
|
"""DatasetRunContextDto — run context with launch audit."""
|
|
|
|
def test_minimal(self):
|
|
from src.schemas.dataset_review_pkg._composites import DatasetRunContextDto
|
|
from src.models.dataset_review import LaunchStatus
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = DatasetRunContextDto(
|
|
run_context_id="rc-1", session_id="s-1",
|
|
dataset_ref="ds1", environment_id="env-1",
|
|
preview_id="p-1", sql_lab_session_ref="sql1",
|
|
effective_filters={}, template_params={},
|
|
approved_mapping_ids=[], semantic_decision_refs=[],
|
|
open_warning_refs=[],
|
|
launch_status=LaunchStatus.SUCCESS,
|
|
created_at=ts,
|
|
)
|
|
assert dto.launch_status == "success"
|
|
|
|
|
|
class TestSessionSummary:
|
|
"""SessionSummary — lightweight session DTO."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.dataset_review import ReadinessState, RecommendedAction, SessionPhase, SessionStatus
|
|
from src.schemas.dataset_review_pkg._composites import SessionSummary
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SessionSummary(
|
|
session_id="s-1", user_id="u-1",
|
|
environment_id="env-1", source_kind="dashboard",
|
|
source_input="42", dataset_ref="ds1",
|
|
readiness_state=ReadinessState.EMPTY,
|
|
recommended_action=RecommendedAction.IMPORT_FROM_SUPERSET,
|
|
status=SessionStatus.ACTIVE,
|
|
current_phase=SessionPhase.INTAKE,
|
|
created_at=ts, updated_at=ts, last_activity_at=ts,
|
|
)
|
|
assert dto.session_id == "s-1"
|
|
|
|
|
|
class TestSessionDetail:
|
|
"""SessionDetail — full session with nested aggregates."""
|
|
|
|
def test_minimal(self):
|
|
from src.models.dataset_review import ReadinessState, RecommendedAction, SessionPhase, SessionStatus
|
|
from src.schemas.dataset_review_pkg._composites import SessionDetail
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
dto = SessionDetail(
|
|
session_id="s-1", user_id="u-1",
|
|
environment_id="env-1", source_kind="dashboard",
|
|
source_input="42", dataset_ref="ds1",
|
|
readiness_state=ReadinessState.EMPTY,
|
|
recommended_action=RecommendedAction.IMPORT_FROM_SUPERSET,
|
|
status=SessionStatus.ACTIVE,
|
|
current_phase=SessionPhase.INTAKE,
|
|
created_at=ts, updated_at=ts, last_activity_at=ts,
|
|
)
|
|
assert dto.collaborators == []
|
|
assert dto.profile is None
|
|
assert dto.findings == []
|
|
assert dto.semantic_sources == []
|
|
assert dto.semantic_fields == []
|
|
|
|
def test_with_nested(self):
|
|
from src.models.dataset_review import (
|
|
BusinessSummarySource, ConfidenceState, ReadinessState, RecommendedAction,
|
|
SessionCollaboratorRole, SessionPhase, SessionStatus,
|
|
)
|
|
from src.schemas.dataset_review_pkg._composites import SessionDetail
|
|
from src.schemas.dataset_review_pkg._dtos import DatasetProfileDto, SessionCollaboratorDto
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
collab = SessionCollaboratorDto(user_id="u-1", role=SessionCollaboratorRole.VIEWER, added_at=ts)
|
|
profile = DatasetProfileDto(
|
|
profile_id="p-1", session_id="s-1", dataset_name="sales",
|
|
business_summary="Sales data",
|
|
business_summary_source=BusinessSummarySource.IMPORTED,
|
|
is_sqllab_view=False, confidence_state=ConfidenceState.CONFIRMED,
|
|
has_blocking_findings=False, has_warning_findings=False,
|
|
manual_summary_locked=False, created_at=ts, updated_at=ts,
|
|
)
|
|
dto = SessionDetail(
|
|
session_id="s-1", user_id="u-1", environment_id="env-1",
|
|
source_kind="dashboard", source_input="42", dataset_ref="ds1",
|
|
readiness_state=ReadinessState.EMPTY,
|
|
recommended_action=RecommendedAction.IMPORT_FROM_SUPERSET,
|
|
status=SessionStatus.ACTIVE, current_phase=SessionPhase.INTAKE,
|
|
created_at=ts, updated_at=ts, last_activity_at=ts,
|
|
collaborators=[collab], profile=profile,
|
|
)
|
|
assert len(dto.collaborators) == 1
|
|
assert dto.profile is not None
|
|
assert dto.profile.dataset_name == "sales"
|
|
|
|
def test_facade_reimport(self):
|
|
"""Verify dataset_review.py facade re-exports work."""
|
|
from src.schemas.dataset_review import SessionDetail, SessionSummary, DatasetProfileDto
|
|
assert SessionDetail is not None
|
|
assert SessionSummary is not None
|
|
assert DatasetProfileDto is not None
|
|
# #endregion Test.Schemas.DatasetReview
|