busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: finalize semantic repair and test updates

Browse Source
This commit is contained in:
busya
2026-03-21 15:07:06 +03:00
parent 1a79589632
commit 07c064bb86
99 changed files with 2484 additions and 985 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
backend/src/services/dataset_review/orchestrator.py
View File

@@ -5,10 +5,10 @@
# @LAYER: Domain
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
# @RELATION: [DEPENDS_ON] ->[SemanticSourceResolver]
# @RELATION: [DEPENDS_ON] ->[ClarificationEngine]
# @RELATION: [DEPENDS_ON] ->[SupersetContextExtractor]
# @RELATION: [DEPENDS_ON] ->[SupersetCompilationAdapter]
# @RELATION: [DEPENDS_ON] ->[TaskManager]
# @RELATION: [CONTAINS] ->[DatasetReviewOrchestrator]
# @PRE: session mutations must execute inside a persisted session boundary scoped to one authenticated user.
# @POST: state transitions are persisted atomically and emit observable progress for long-running steps.
# @SIDE_EFFECT: creates task records, updates session aggregates, triggers upstream Superset calls, persists audit artifacts.
@@ -158,8 +158,8 @@ class LaunchDatasetResult:
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
# @RELATION: [DEPENDS_ON] ->[SupersetContextExtractor]
# @RELATION: [DEPENDS_ON] ->[TaskManager]
# @RELATION: [DEPENDS_ON] ->[SessionRepo]
# @RELATION: [DEPENDS_ON] ->[ConfigManager]
# @RELATION: [DEPENDS_ON] ->[SemanticSourceResolver]
# @PRE: constructor dependencies are valid and tied to the current request/task scope.
# @POST: orchestrator instance can execute session-scoped mutations for one authenticated user.
# @SIDE_EFFECT: downstream operations may persist session/profile/finding state and enqueue background tasks.
@@ -169,8 +169,13 @@ class DatasetReviewOrchestrator:
# [DEF:DatasetReviewOrchestrator_init:Function]
# @COMPLEXITY: 3
# @PURPOSE: Bind repository, config, and task dependencies required by the orchestration boundary.
# @RELATION: [DEPENDS_ON] ->[SessionRepo]
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
# @RELATION: [DEPENDS_ON] ->[ConfigManager]
# @RELATION: [DEPENDS_ON] ->[TaskManager]
# @RELATION: [DEPENDS_ON] ->[SemanticSourceResolver]
# @PRE: repository/config_manager are valid collaborators for the current request scope.
# @POST: Instance holds collaborator references used by start/preview/launch orchestration methods.
# @SIDE_EFFECT: Stores dependency references for later session lifecycle operations.
def __init__(
self,
repository: DatasetReviewSessionRepository,
@@ -188,9 +193,9 @@ class DatasetReviewOrchestrator:
# [DEF:start_session:Function]
# @COMPLEXITY: 5
# @PURPOSE: Initialize a new session from a Superset link or dataset selection and trigger context recovery.
# @RELATION: [DEPENDS_ON] ->[SessionRepo]
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
# @RELATION: [CALLS] ->[SupersetContextExtractor.parse_superset_link]
# @RELATION: [CALLS] ->[create_task]
# @RELATION: [CALLS] ->[TaskManager.create_task]
# @PRE: source input is non-empty and environment is accessible.
# @POST: session exists in persisted storage with intake/recovery state and task linkage when async work is required.
# @SIDE_EFFECT: persists session and may enqueue recovery task.
@@ -1112,7 +1117,7 @@ class DatasetReviewOrchestrator:
# [DEF:_enqueue_recovery_task:Function]
# @COMPLEXITY: 4
# @PURPOSE: Link session start to observable async recovery when task infrastructure is available.
# @RELATION: [CALLS] ->[create_task]
# @RELATION: [CALLS] ->[TaskManager.create_task]
# @PRE: session is already persisted.
# @POST: returns task identifier when a task could be enqueued, otherwise None.
# @SIDE_EFFECT: may create one background task for progressive recovery.

169
backend/src/services/dataset_review/repositories/__tests__/test_session_repository.py
View File

@@ -16,34 +16,46 @@ from src.models.dataset_review import (
ReadinessState,
RecommendedAction,
SessionCollaborator,
SessionCollaboratorRole
SessionCollaboratorRole,
)
from src.services.dataset_review.repositories.session_repository import (
DatasetReviewSessionRepository,
)
from src.services.dataset_review.repositories.session_repository import DatasetReviewSessionRepository
# [DEF:SessionRepositoryTests:Module]
# @RELATION: BELONGS_TO -> SrcRoot
# @COMPLEXITY: 2
# @PURPOSE: Unit tests for DatasetReviewSessionRepository.
@pytest.fixture
def db_session():
# [DEF:db_session:Function]
# @COMPLEXITY: 2
# @PURPOSE: Build isolated in-memory SQLAlchemy session seeded with baseline user/environment rows.
# @RELATION: BINDS_TO -> [SessionRepositoryTests]
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# Create test data
user = User(id="user1", username="testuser", email="test@example.com", password_hash="pw")
env = Environment(id="env1", name="Prod", url="http://superset", credentials_id="cred1")
user = User(
id="user1", username="testuser", email="test@example.com", password_hash="pw"
)
env = Environment(
id="env1", name="Prod", url="http://superset", credentials_id="cred1"
)
session.add_all([user, env])
session.commit()
yield session
session.close()
# [/DEF:db_session:Function]
# [DEF:test_create_session:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_create_session(db_session):
@@ -54,126 +66,156 @@ def test_create_session(db_session):
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1"
dataset_ref="dataset1",
)
repo.create_session(session)
assert session.session_id is not None
loaded = db_session.query(DatasetReviewSession).filter_by(session_id=session.session_id).first()
loaded = (
db_session.query(DatasetReviewSession)
.filter_by(session_id=session.session_id)
.first()
)
assert loaded.user_id == "user1"
# [/DEF:test_create_session:Function]
# [DEF:test_load_session_detail_ownership:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_load_session_detail_ownership(db_session):
# @PURPOSE: Verify ownership enforcement in detail loading.
repo = DatasetReviewSessionRepository(db_session)
session = DatasetReviewSession(
user_id="user1", environment_id="env1", source_kind="superset_link",
source_input="http://link", dataset_ref="dataset1"
user_id="user1",
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1",
)
repo.create_session(session)
# Correct user
loaded = repo.load_session_detail(session.session_id, "user1")
assert loaded is not None
# Wrong user
loaded_wrong = repo.load_session_detail(session.session_id, "wrong_user")
assert loaded_wrong is None
# [/DEF:test_load_session_detail_ownership:Function]
# [DEF:test_load_session_detail_collaborator:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_load_session_detail_collaborator(db_session):
# @PURPOSE: Verify collaborator access in detail loading.
repo = DatasetReviewSessionRepository(db_session)
session = DatasetReviewSession(
user_id="user1", environment_id="env1", source_kind="superset_link",
source_input="http://link", dataset_ref="dataset1"
user_id="user1",
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1",
)
repo.create_session(session)
# Add collaborator
collab_user = User(id="collab1", username="collab", email="c@e.com", password_hash="p")
collab_user = User(
id="collab1", username="collab", email="c@e.com", password_hash="p"
)
db_session.add(collab_user)
collaborator = SessionCollaborator(
session_id=session.session_id,
user_id="collab1",
role=SessionCollaboratorRole.REVIEWER
role=SessionCollaboratorRole.REVIEWER,
)
db_session.add(collaborator)
db_session.commit()
# Collaborator access
loaded = repo.load_session_detail(session.session_id, "collab1")
assert loaded is not None
assert loaded.session_id == session.session_id
# [/DEF:test_load_session_detail_collaborator:Function]
# [DEF:test_save_preview_marks_stale:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_save_preview_marks_stale(db_session):
# @PURPOSE: Verify that saving a new preview marks old ones as stale.
repo = DatasetReviewSessionRepository(db_session)
session = DatasetReviewSession(
user_id="user1", environment_id="env1", source_kind="superset_link",
source_input="http://link", dataset_ref="dataset1"
user_id="user1",
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1",
)
repo.create_session(session)
p1 = CompiledPreview(session_id=session.session_id, preview_status="ready", preview_fingerprint="f1")
p1 = CompiledPreview(
session_id=session.session_id, preview_status="ready", preview_fingerprint="f1"
)
repo.save_preview(session.session_id, "user1", p1)
p2 = CompiledPreview(session_id=session.session_id, preview_status="ready", preview_fingerprint="f2")
p2 = CompiledPreview(
session_id=session.session_id, preview_status="ready", preview_fingerprint="f2"
)
repo.save_preview(session.session_id, "user1", p2)
db_session.refresh(p1)
assert p1.preview_status == "stale"
assert p2.preview_status == "ready"
assert session.last_preview_id == p2.preview_id
# [/DEF:test_save_preview_marks_stale:Function]
# [DEF:test_save_profile_and_findings:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_save_profile_and_findings(db_session):
# @PURPOSE: Verify persistence of profile and findings.
repo = DatasetReviewSessionRepository(db_session)
session = DatasetReviewSession(
user_id="user1", environment_id="env1", source_kind="superset_link",
source_input="http://link", dataset_ref="dataset1"
user_id="user1",
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1",
)
repo.create_session(session)
profile = DatasetProfile(
session_id=session.session_id,
dataset_name="Test DS",
business_summary="Summary",
business_summary_source=BusinessSummarySource.INFERRED,
confidence_state=ConfidenceState.UNRESOLVED
confidence_state=ConfidenceState.UNRESOLVED,
)
finding = ValidationFinding(
session_id=session.session_id,
area=FindingArea.SOURCE_INTAKE,
severity=FindingSeverity.BLOCKING,
code="ERR1",
title="Error",
message="Failure"
message="Failure",
)
repo.save_profile_and_findings(session.session_id, "user1", profile, [finding])
updated_session = repo.load_session_detail(session.session_id, "user1")
assert updated_session.profile.dataset_name == "Test DS"
assert len(updated_session.findings) == 1
assert updated_session.findings[0].code == "ERR1"
# Verify removal of old findings
new_finding = ValidationFinding(
session_id=session.session_id,
@@ -181,29 +223,34 @@ def test_save_profile_and_findings(db_session):
severity=FindingSeverity.WARNING,
code="WARN1",
title="Warning",
message="Something"
message="Something",
)
repo.save_profile_and_findings(session.session_id, "user1", profile, [new_finding])
db_session.expire_all()
final_session = repo.load_session_detail(session.session_id, "user1")
assert len(final_session.findings) == 1
assert final_session.findings[0].code == "WARN1"
# [/DEF:test_save_profile_and_findings:Function]
# [DEF:test_save_run_context:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_save_run_context(db_session):
# @PURPOSE: Verify saving of run context.
repo = DatasetReviewSessionRepository(db_session)
session = DatasetReviewSession(
user_id="user1", environment_id="env1", source_kind="superset_link",
source_input="http://link", dataset_ref="dataset1"
user_id="user1",
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1",
)
repo.create_session(session)
rc = DatasetRunContext(
session_id=session.session_id,
dataset_ref="ds1",
@@ -215,28 +262,50 @@ def test_save_run_context(db_session):
approved_mapping_ids=[],
semantic_decision_refs=[],
open_warning_refs=[],
launch_status="success"
launch_status="success",
)
repo.save_run_context(session.session_id, "user1", rc)
assert session.last_run_context_id == rc.run_context_id
# [/DEF:test_save_run_context:Function]
# [DEF:test_list_sessions_for_user:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_list_sessions_for_user(db_session):
# @PURPOSE: Verify listing of sessions by user.
repo = DatasetReviewSessionRepository(db_session)
s1 = DatasetReviewSession(user_id="user1", environment_id="env1", source_kind="k", source_input="i", dataset_ref="r1")
s2 = DatasetReviewSession(user_id="user1", environment_id="env1", source_kind="k", source_input="i", dataset_ref="r2")
s3 = DatasetReviewSession(user_id="other", environment_id="env1", source_kind="k", source_input="i", dataset_ref="r3")
s1 = DatasetReviewSession(
user_id="user1",
environment_id="env1",
source_kind="k",
source_input="i",
dataset_ref="r1",
)
s2 = DatasetReviewSession(
user_id="user1",
environment_id="env1",
source_kind="k",
source_input="i",
dataset_ref="r2",
)
s3 = DatasetReviewSession(
user_id="other",
environment_id="env1",
source_kind="k",
source_input="i",
dataset_ref="r3",
)
db_session.add_all([s1, s2, s3])
db_session.commit()
sessions = repo.list_sessions_for_user("user1")
assert len(sessions) == 2
assert all(s.user_id == "user1" for s in sessions)
# [/DEF:SessionRepositoryTests:Module]# [/DEF:test_list_sessions_for_user:Function]
# [/DEF:test_list_sessions_for_user:Function]
# [/DEF:SessionRepositoryTests:Module]

5
backend/src/services/dataset_review/repositories/session_repository.py
View File

@@ -34,13 +34,14 @@ from src.core.logger import belief_scope, logger
from src.services.dataset_review.event_logger import SessionEventLogger
# [DEF:SessionRepo:Class]
# [DEF:DatasetReviewSessionRepository:Class]
# @COMPLEXITY: 4
# @PURPOSE: Enforce ownership-scoped persistence and retrieval for dataset review session aggregates.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [DatasetProfile]
# @RELATION: [DEPENDS_ON] -> [ValidationFinding]
# @RELATION: [DEPENDS_ON] -> [CompiledPreview]
# @RELATION: [DEPENDS_ON] -> [SessionEventLogger]
# @PRE: constructor receives a live SQLAlchemy session and callers provide authenticated user scope for guarded reads and writes.
# @POST: repository methods return ownership-scoped aggregates or persisted child records without changing domain meaning.
# @SIDE_EFFECT: mutates and queries the persistence layer through the injected database session.
@@ -409,6 +410,6 @@ class DatasetReviewSessionRepository:
# [/DEF:list_user_sess:Function]
# [/DEF:SessionRepo:Class]
# [/DEF:DatasetReviewSessionRepository:Class]
# [/DEF:DatasetReviewSessionRepository:Module]