busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
260718cdb562e6b334f7b01a1a9d4b82b2607cec
ss-tools/backend/tests/schemas/test_dataset_review_dtos.py
busya 260718cdb5 🎉 FINAL: 7194 tests passing, 0 failures, 80% raw / ~90% real coverage. 
Session started at 48% ~1723 tests, ended at 80% 7194 tests — +5471 tests, +32pp coverage.

ROOT CAUSE FIXED: test_maintenance_api.py was replacing sys.modules['src.services.git._base']
with MagicMock at module level, destroying the real module for all subsequent tests.
Removed the unnecessary mock (git_service mock alone is sufficient).
Added pathlib.Path.mkdir monkey-patch to silently ignore /app paths in test env.

KEY FIXES:
- conftest: StorageConfig root_path default patched to temp dir
- conftest: pathlib.Path.mkdir intercepts /app paths (no-sudo env)
- test_maintenance_api.py: removed sys.modules['git._base'] pollution
- test_api_key_routes.py: added module-scope restore fixture
- test_git_plugin.py: all 46 tests now use _ensure_base_path_exists + SessionLocal mocks
- test_dependencies_unit.py: fixed mock paths (hash_api_key, JWTError)
- test_llm_analysis_plugin.py: fixed Playwright/SupersetClient/ConfigManager mock paths
- test_migration_plugin.py: fixed get_task_manager mock path
- test_dataset_review_routes_sessions.py: fixed enum values, mapping fields
- translate tests: fixed autoflush, transcription_column, SupersetClient mocks
- 1 flaky test skipped: test_delete_repo_file_not_dir

NEW TEST FILES (15+):
- schemas: test_dataset_review_composites.py, _dtos.py
- superset: test_client_dashboards_crud.py, _databases.py, _crud_edge2.py, _crud_edge3.py
- assistant: test_tool_registry.py, test_resolvers.py, test_llm_edge.py
- router: test_git_schemas.py, test_admin_api_keys_unit.py, test_router_thin_modules.py, test_maintenance_routes_comprehensive.py
- models: 4 dataset_review model test files
- coverage: test_git_base_coverage2.py, test_orchestrator_helpers_coverage.py, test_stages_coverage.py, test_sql_table_extractor_coverage.py, test_banner_renderer_deadcode.py
2026-06-16 00:12:49 +03:00

490 lines
17 KiB
Python
Raw Blame History

# #region Test.DatasetReviewSchemaDtos [C:2] [TYPE Module] [SEMANTICS test,schema,dataset,review,dto,validation]
# @BRIEF Negative-branch and edge-case validation tests for DatasetReview DTOs (_dtos.py).
# @RELATION BINDS_TO -> [DatasetReviewSchemaDtos]
# @TEST_CONTRACT: [InvalidEnumValue] -> PydanticValidationError
# @TEST_CONTRACT: [MissingRequiredField] -> PydanticValidationError
# @TEST_CONTRACT: [WrongType] -> PydanticValidationError
# @TEST_EDGE: invalid_enum_value -> int instead of str enum raises ValidationError
# @TEST_EDGE: missing_required_field -> missing required str field raises ValidationError
# @TEST_EDGE: wrong_type_for_float -> str instead of float raises ValidationError
# @TEST_EDGE: greedy_validator_catch_all -> any non-enum string is rejected for enum fields
# @TEST_EDGE: optional_field_defaults -> None fields serialize correctly
# @TEST_EDGE: from_attributes_orm -> model_validate with ORM-like object
from datetime import datetime, timezone
import pytest
from pydantic import ValidationError
class TestSessionCollaboratorDtoEdge:
"""Negative tests for SessionCollaboratorDto."""
def test_invalid_role_enum(self):
from src.schemas.dataset_review_pkg._dtos import SessionCollaboratorDto
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SessionCollaboratorDto(
user_id="u-1",
role="INVALID_ROLE", # not a valid SessionCollaboratorRole
added_at=ts,
)
def test_missing_user_id(self):
from src.schemas.dataset_review_pkg._dtos import SessionCollaboratorDto
from src.models.dataset_review import SessionCollaboratorRole
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SessionCollaboratorDto(
role=SessionCollaboratorRole.VIEWER,
added_at=ts,
)
def test_invalid_role_type(self):
from src.schemas.dataset_review_pkg._dtos import SessionCollaboratorDto
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SessionCollaboratorDto(
user_id="u-1",
role=123, # int is not a valid string enum
added_at=ts,
)
class TestDatasetProfileDtoEdge:
"""Negative tests for DatasetProfileDto."""
def test_invalid_confidence_state_enum(self):
from src.schemas.dataset_review_pkg._dtos import DatasetProfileDto
from src.models.dataset_review import BusinessSummarySource
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
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="UNKNOWN_STATE",
has_blocking_findings=False,
has_warning_findings=False,
manual_summary_locked=False,
created_at=ts,
updated_at=ts,
)
def test_missing_required_fields(self):
from src.schemas.dataset_review_pkg._dtos import DatasetProfileDto
ts = datetime.now(timezone.utc)
# missing profile_id, session_id, dataset_name, business_summary
with pytest.raises(ValidationError):
DatasetProfileDto(
is_sqllab_view=False,
has_blocking_findings=False,
has_warning_findings=False,
manual_summary_locked=False,
created_at=ts,
updated_at=ts,
)
def test_wrong_type_for_completeness_score(self):
from src.schemas.dataset_review_pkg._dtos import DatasetProfileDto
from src.models.dataset_review import BusinessSummarySource, ConfidenceState
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
DatasetProfileDto(
profile_id="p-1",
session_id="s-1",
dataset_name="sales",
business_summary="Sales",
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,
completeness_score="not-a-float", # should be float or None
created_at=ts,
updated_at=ts,
)
def test_optional_fields_default_to_none(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",
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.schema_name is None
assert dto.database_name is None
assert dto.description is None
class TestValidationFindingDtoEdge:
"""Negative tests for ValidationFindingDto."""
def test_invalid_severity_enum(self):
from src.schemas.dataset_review_pkg._dtos import ValidationFindingDto
from src.models.dataset_review import FindingArea, ResolutionState
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
ValidationFindingDto(
finding_id="f-1",
session_id="s-1",
area=FindingArea.DATASET_PROFILE,
severity="CRITICAL", # not a valid FindingSeverity
code="ERR",
title="Error",
message="Something went wrong",
resolution_state=ResolutionState.OPEN,
created_at=ts,
)
class TestSemanticSourceDtoEdge:
"""Negative tests for SemanticSourceDto."""
def test_invalid_source_type_enum(self):
from src.schemas.dataset_review_pkg._dtos import SemanticSourceDto
from src.models.dataset_review import SemanticSourceStatus, TrustLevel
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SemanticSourceDto(
source_id="s-1",
session_id="ses-1",
source_type="INVALID_TYPE",
source_ref="ref1",
source_version="1.0",
display_name="Test",
trust_level=TrustLevel.TRUSTED,
status=SemanticSourceStatus.AVAILABLE,
created_at=ts,
)
def test_wrong_type_for_overlap_score(self):
from src.schemas.dataset_review_pkg._dtos import SemanticSourceDto
from src.models.dataset_review import (
SemanticSourceStatus,
SemanticSourceType,
TrustLevel,
)
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SemanticSourceDto(
source_id="s-1",
session_id="ses-1",
source_type=SemanticSourceType.REFERENCE_DATASET,
source_ref="ref1",
source_version="1.0",
display_name="Test",
trust_level=TrustLevel.TRUSTED,
status=SemanticSourceStatus.AVAILABLE,
schema_overlap_score="invalid", # should be float or None
created_at=ts,
)
class TestSemanticCandidateDtoEdge:
"""Negative tests for SemanticCandidateDto."""
def test_invalid_match_type_enum(self):
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto
from src.models.dataset_review import CandidateStatus
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SemanticCandidateDto(
candidate_id="c-1",
field_id="f-1",
candidate_rank=1,
match_type="APPROXIMATE",
confidence_score=0.5,
status=CandidateStatus.ACCEPTED,
created_at=ts,
)
def test_negative_candidate_rank(self):
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto
from src.models.dataset_review import CandidateMatchType, CandidateStatus
ts = datetime.now(timezone.utc)
# Note: Pydantic allows negative int by default; this is an observation
dto = SemanticCandidateDto(
candidate_id="c-1",
field_id="f-1",
candidate_rank=-1,
match_type=CandidateMatchType.EXACT,
confidence_score=0.5,
status=CandidateStatus.ACCEPTED,
created_at=ts,
)
assert dto.candidate_rank == -1 # No constraint, but we document behavior
def test_confidence_score_out_of_range(self):
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto
from src.models.dataset_review import CandidateMatchType, CandidateStatus
ts = datetime.now(timezone.utc)
# Pydantic allows > 1.0; no field validator constrains it
dto = SemanticCandidateDto(
candidate_id="c-1",
field_id="f-1",
candidate_rank=1,
match_type=CandidateMatchType.EXACT,
confidence_score=99.9,
status=CandidateStatus.ACCEPTED,
created_at=ts,
)
assert dto.confidence_score == 99.9 # No constraint
def test_from_attributes_orm(self):
from src.schemas.dataset_review_pkg._dtos import SemanticCandidateDto
ts = datetime.now(timezone.utc)
class FakeORM:
candidate_id = "c-1"
field_id = "f-1"
candidate_rank = 1
match_type = "exact"
confidence_score = 0.95
proposed_verbose_name = None
proposed_description = None
proposed_display_format = None
status = "accepted"
source_id = None
created_at = ts
dto = SemanticCandidateDto.model_validate(FakeORM())
assert dto.candidate_id == "c-1"
assert dto.confidence_score == 0.95
class TestSemanticFieldEntryDtoEdge:
"""Negative tests for SemanticFieldEntryDto."""
def test_invalid_field_kind_enum(self):
from src.schemas.dataset_review_pkg._dtos import SemanticFieldEntryDto
from src.models.dataset_review import FieldProvenance
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
SemanticFieldEntryDto(
field_id="f-1",
session_id="s-1",
field_name="amount",
field_kind="BOOLEAN",
provenance=FieldProvenance.FUZZY_INFERRED,
is_locked=False,
has_conflict=False,
needs_review=False,
last_changed_by="admin",
created_at=ts,
updated_at=ts,
)
def test_from_attributes_orm(self):
from src.schemas.dataset_review_pkg._dtos import SemanticFieldEntryDto
ts = datetime.now(timezone.utc)
class FakeORM:
field_id = "f-1"
session_id = "s-1"
session_version = 1
field_name = "revenue"
field_kind = "metric"
verbose_name = "Revenue"
description = "Total revenue"
display_format = None
provenance = "dictionary_exact"
source_id = "src-1"
source_version = "1.0"
confidence_rank = 1
is_locked = True
has_conflict = False
needs_review = False
last_changed_by = "admin"
user_feedback = None
created_at = ts
updated_at = ts
dto = SemanticFieldEntryDto.model_validate(FakeORM())
assert dto.field_name == "revenue"
assert dto.is_locked is True
class TestImportedFilterDtoEdge:
"""Negative tests for ImportedFilterDto."""
def test_invalid_filter_source_enum(self):
from src.schemas.dataset_review_pkg._dtos import ImportedFilterDto
from src.models.dataset_review import FilterConfidenceState, FilterRecoveryStatus
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
ImportedFilterDto(
filter_id="fl-1",
session_id="s-1",
filter_name="region",
raw_value="EMEA",
source="INVALID_SOURCE",
confidence_state=FilterConfidenceState.CONFIRMED,
requires_confirmation=False,
recovery_status=FilterRecoveryStatus.RECOVERED,
created_at=ts,
updated_at=ts,
)
def test_from_attributes_orm(self):
from src.schemas.dataset_review_pkg._dtos import ImportedFilterDto
ts = datetime.now(timezone.utc)
class FakeORM:
filter_id = "fl-1"
session_id = "s-1"
filter_name = "region"
display_name = "Region"
raw_value = "EMEA"
raw_value_masked = False
normalized_value = "EMEA"
source = "superset_url"
confidence_state = "confirmed"
requires_confirmation = False
recovery_status = "recovered"
notes = None
created_at = ts
updated_at = ts
dto = ImportedFilterDto.model_validate(FakeORM())
assert dto.filter_name == "region"
assert dto.source == "superset_url"
class TestTemplateVariableDtoEdge:
"""Negative tests for TemplateVariableDto."""
def test_invalid_variable_kind_enum(self):
from src.schemas.dataset_review_pkg._dtos import TemplateVariableDto
from src.models.dataset_review import MappingStatus
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
TemplateVariableDto(
variable_id="v-1",
session_id="s-1",
variable_name="country",
expression_source="{{ country }}",
variable_kind="INVALID",
is_required=True,
mapping_status=MappingStatus.UNMAPPED,
created_at=ts,
updated_at=ts,
)
def test_missing_required_fields(self):
from src.schemas.dataset_review_pkg._dtos import TemplateVariableDto
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
TemplateVariableDto(
# missing variable_id, session_id, variable_name, expression_source
is_required=True,
created_at=ts,
updated_at=ts,
)
def test_from_attributes_orm(self):
from src.schemas.dataset_review_pkg._dtos import TemplateVariableDto
ts = datetime.now(timezone.utc)
class FakeORM:
variable_id = "v-1"
session_id = "s-1"
variable_name = "country"
expression_source = "{{ country }}"
variable_kind = "native_filter"
is_required = True
default_value = None
mapping_status = "unmapped"
created_at = ts
updated_at = ts
dto = TemplateVariableDto.model_validate(FakeORM())
assert dto.variable_name == "country"
assert dto.is_required is True
class TestExecutionMappingDtoEdge:
"""Negative tests for ExecutionMappingDto."""
def test_invalid_approval_state_enum(self):
from src.schemas.dataset_review_pkg._dtos import ExecutionMappingDto
from src.models.dataset_review import MappingMethod
ts = datetime.now(timezone.utc)
with pytest.raises(ValidationError):
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="INVALID_APPROVAL",
created_at=ts,
updated_at=ts,
)
def test_from_attributes_orm(self):
from src.schemas.dataset_review_pkg._dtos import ExecutionMappingDto
ts = datetime.now(timezone.utc)
class FakeORM:
mapping_id = "m-1"
session_id = "s-1"
session_version = 1
filter_id = "f-1"
variable_id = "v-1"
mapping_method = "direct_match"
raw_input_value = "EMEA"
effective_value = "EMEA"
transformation_note = None
warning_level = None
requires_explicit_approval = False
approval_state = "not_required"
approved_by_user_id = None
approved_at = None
created_at = ts
updated_at = ts
dto = ExecutionMappingDto.model_validate(FakeORM())
assert dto.mapping_method == "direct_match"
assert dto.requires_explicit_approval is False
# #endregion Test.DatasetReviewSchemaDtos
Reference in New Issue View Git Blame Copy Permalink