busya
f75c15dbc6
- 10 translate plugin test files (100% coverage on 12 modules) - assistant/handler tools: 85+ tests covering dispatch, registry, resolvers, routes, llm_planner, 13 tool handlers - clean release: artifact_catalog_loader, mappers, approval, publication tests - API routes: translate_helpers, validation_service extensions, datasets to 100% - notifications: providers/service tests - services: profile_preference_service - docs/orthogonal-test-report.md — full speckit.tests audit - Fixes: 3 git_base async mock failures, 4 assistant handler permission-check patches - .gitignore: coverage artifacts
112 lines
4.6 KiB
Python
112 lines
4.6 KiB
Python
# #region Test.OrchestratorValidation [C:3] [TYPE Module] [SEMANTICS test, translate, validation, preconditions]
|
|
# @BRIEF Tests for orchestrator_validation.py — validate_job_preconditions.
|
|
# @RELATION BINDS_TO -> [validate_job_preconditions]
|
|
# @TEST_CONTRACT: validate_job_preconditions -> None | raises ValueError on precondition failure
|
|
# @TEST_EDGE: draft_status_fails
|
|
# @TEST_EDGE: missing_target_table
|
|
# @TEST_EDGE: missing_provider_id
|
|
# @TEST_EDGE: missing_translation_column
|
|
# @TEST_EDGE: no_accepted_preview_session
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
|
|
|
|
import os
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing")
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.plugins.translate.orchestrator_validation import validate_job_preconditions
|
|
|
|
|
|
class TestValidateJobPreconditions:
|
|
"""validate_job_preconditions — precondition checks."""
|
|
|
|
def _make_job(self, overrides: dict | None = None) -> MagicMock:
|
|
"""Create a mock job with valid defaults."""
|
|
job = MagicMock()
|
|
job.id = "job-1"
|
|
job.status = "READY"
|
|
job.target_table = "target_table"
|
|
job.provider_id = "provider-1"
|
|
job.translation_column = "name"
|
|
job.source_datasource_id = None
|
|
if overrides:
|
|
for k, v in overrides.items():
|
|
setattr(job, k, v)
|
|
return job
|
|
|
|
def test_ready_passes_with_source_datasource(self):
|
|
"""READY job with source datasource passes without preview session."""
|
|
job = self._make_job({"source_datasource_id": "ds-1"})
|
|
db = MagicMock()
|
|
# Should not query preview sessions
|
|
validate_job_preconditions(job, db)
|
|
db.query.assert_not_called()
|
|
|
|
def test_ready_no_datasource_needs_preview(self):
|
|
"""READY job without datasource and no accepted preview fails."""
|
|
job = self._make_job({"source_datasource_id": None})
|
|
db = MagicMock()
|
|
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
|
|
with pytest.raises(ValueError, match="no accepted preview session"):
|
|
validate_job_preconditions(job, db)
|
|
|
|
def test_ready_no_datasource_with_preview_passes(self):
|
|
"""READY job without datasource but with accepted preview passes."""
|
|
job = self._make_job({"source_datasource_id": None})
|
|
db = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_session.status = "APPLIED"
|
|
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = mock_session
|
|
validate_job_preconditions(job, db)
|
|
|
|
def test_draft_status_fails(self):
|
|
"""DRAFT job raises ValueError."""
|
|
job = self._make_job({"status": "DRAFT"})
|
|
db = MagicMock()
|
|
with pytest.raises(ValueError, match="Cannot run job.*DRAFT"):
|
|
validate_job_preconditions(job, db)
|
|
|
|
def test_missing_target_table(self):
|
|
"""Job with no target_table raises ValueError."""
|
|
job = self._make_job({"target_table": None})
|
|
db = MagicMock()
|
|
with pytest.raises(ValueError, match="no target table"):
|
|
validate_job_preconditions(job, db)
|
|
|
|
def test_missing_provider_id(self):
|
|
"""Job with no provider_id raises ValueError."""
|
|
job = self._make_job({"provider_id": None})
|
|
db = MagicMock()
|
|
with pytest.raises(ValueError, match="no LLM provider"):
|
|
validate_job_preconditions(job, db)
|
|
|
|
def test_missing_translation_column(self):
|
|
"""Job with no translation_column raises ValueError."""
|
|
job = self._make_job({"translation_column": None})
|
|
db = MagicMock()
|
|
with pytest.raises(ValueError, match="no translation column"):
|
|
validate_job_preconditions(job, db)
|
|
|
|
def test_scheduled_run_skips_preview_check(self):
|
|
"""Scheduled run (is_scheduled=True) skips preview session check."""
|
|
job = self._make_job({"source_datasource_id": None})
|
|
db = MagicMock()
|
|
# Would fail without is_scheduled=True, but should pass with it
|
|
validate_job_preconditions(job, db, is_scheduled=True)
|
|
db.query.assert_not_called()
|
|
|
|
def test_validation_order_draft_checked_first(self):
|
|
"""DRAFT is checked before target_table."""
|
|
job = self._make_job({"status": "DRAFT", "target_table": None})
|
|
db = MagicMock()
|
|
with pytest.raises(ValueError, match="DRAFT"):
|
|
validate_job_preconditions(job, db)
|
|
# #endregion Test.OrchestratorValidation
|