test(backend): raise coverage to 95%+ statements and branches (97.8%/95.0%)

- ~60 new/extended test files across api, core, plugins, services, schemas:
  routes, superset clients, task_manager, lineage, git, translate,
  dashboard-testing, load-testing, migration, llm_analysis, scheduler, ssl
- .coveragerc: enable branch coverage; exclude src/__tests__ (test files)
  and src/scripts (CLI/ops tools) from the denominator
- bug fixes found while testing:
  * settings: PUT /settings/reports registered under duplicated prefix
  * schemas/lineage: FleetReportDTO missing run_status (route always 500)
  * dashboard_testing/baseline_inheritance: visual entry read wrong field
  * superset_client/_databases: logger extra name shadowed LogRecord attr
  * routes/datasets: _yaml_string_paths recursion without yield from
  * translate/sql_generator: restore explicit-type timestamp contract
  * baseline_catalog: remove unreachable dashboard_id fallback
- conftest fixes: pytest_plugins to rootdir conftest (pytest 9), test
  filename collision, TMPDIR-safe integration fixtures
This commit is contained in:
2026-08-19 17:14:32 +03:00
parent dd9df0fc5e
commit 488a8f349b
147 changed files with 32459 additions and 57 deletions

View File

@@ -252,5 +252,173 @@ class TestNormalizeTaskReport:
report = normalize_task_report(task)
assert report.details is not None
assert report.details["result"] == {"migrated": 5, "errors": 0}
def test_include_result_false_omits_result(self):
"""List projection (include_result=False) never embeds task.result."""
from src.services.reports.normalizer import normalize_task_report
task = _make_task(result={"migrated": 5, "errors": 0})
report = normalize_task_report(task, include_result=False)
assert report.details is not None
assert "result" not in report.details
assert report.details["profile"]["display_label"] == "Migration"
# #endregion Test.ReportNormalizer.TestNormalizeTaskReport
# #region Test.ReportNormalizer.TestExtractErrorContextLogFallback [C:2] [TYPE Function]
# @BRIEF Test extract_error_context log-scan fallback (reversed logs, case-insensitive levels).
class TestExtractErrorContextLogFallback:
def test_no_matching_error_log_uses_default(self):
from src.services.reports.normalizer import extract_error_context
from src.models.report import ReportStatus
from src.core.task_manager.models import LogEntry
task = _make_task(
result={"other": "data"},
logs=[LogEntry(level="INFO", message="informational only")],
)
ctx = extract_error_context(task, ReportStatus.FAILED)
assert ctx is not None
assert ctx.message == "Not provided"
def test_error_log_level_is_case_insensitive(self):
from src.services.reports.normalizer import extract_error_context
from src.models.report import ReportStatus
from src.core.task_manager.models import LogEntry
task = _make_task(
result={"other": "data"},
logs=[LogEntry(level="Error", message="cased error")],
)
ctx = extract_error_context(task, ReportStatus.FAILED)
assert ctx is not None
assert ctx.message == "cased error"
def test_last_error_log_wins_in_reverse_order(self):
from src.services.reports.normalizer import extract_error_context
from src.models.report import ReportStatus
from src.core.task_manager.models import LogEntry
task = _make_task(
result={"other": "data"},
logs=[
LogEntry(level="ERROR", message="first error"),
LogEntry(level="INFO", message="noise"),
LogEntry(level="ERROR", message="last error"),
],
)
ctx = extract_error_context(task, ReportStatus.FAILED)
assert ctx is not None
assert ctx.message == "last error"
def test_blank_error_message_falls_through_to_logs(self):
from src.services.reports.normalizer import extract_error_context
from src.models.report import ReportStatus
from src.core.task_manager.models import LogEntry
task = _make_task(
result={"error": {"message": "", "code": "E1"}},
logs=[LogEntry(level="ERROR", message="log fallback hit")],
)
ctx = extract_error_context(task, ReportStatus.FAILED)
assert ctx is not None
assert ctx.message == "log fallback hit"
assert ctx.code == "E1"
def test_error_object_actions_strip_blank_entries(self):
from src.services.reports.normalizer import extract_error_context
from src.models.report import ReportStatus
task = _make_task(
result={"error": {"message": "m", "next_actions": [" ", "Fix it", ""]}}
)
ctx = extract_error_context(task, ReportStatus.PARTIAL)
assert ctx is not None
assert ctx.next_actions == ["Fix it"]
# #endregion Test.ReportNormalizer.TestExtractErrorContextLogFallback
# #region Test.ReportNormalizer.TestNormalizeTaskReportLlmValidation [C:2] [TYPE Function]
# @BRIEF Test llm_dashboard_validation success-with-problems override to FAILED/PARTIAL.
class TestNormalizeTaskReportLlmValidation:
def _llm_task(self, **result_overrides):
from src.core.task_manager.models import TaskStatus
result = {"fail_count": 0, "warn_count": 0, "status": "ok"}
result.update(result_overrides)
return _make_task(
plugin_id="llm_dashboard_validation",
status=TaskStatus.SUCCESS,
result=result,
)
def test_fail_count_downgrades_to_failed(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task(fail_count=2))
assert report.status == ReportStatus.FAILED
def test_raw_status_fail_downgrades_to_failed(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task(status="FAIL"))
assert report.status == ReportStatus.FAILED
def test_warn_count_downgrades_to_partial(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task(warn_count=3))
assert report.status == ReportStatus.PARTIAL
def test_raw_status_warn_downgrades_to_partial(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task(status="WARN"))
assert report.status == ReportStatus.PARTIAL
def test_benign_result_stays_success(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task())
assert report.status == ReportStatus.SUCCESS
def test_non_int_counts_stay_success(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task(fail_count="many", warn_count="many"))
assert report.status == ReportStatus.SUCCESS
def test_none_result_stays_success(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
from src.core.task_manager.models import TaskStatus
task = _make_task(
plugin_id="llm_dashboard_validation",
status=TaskStatus.SUCCESS,
result=None,
)
report = normalize_task_report(task)
assert report.status == ReportStatus.SUCCESS
def test_fail_count_takes_priority_over_warn_count(self):
from src.services.reports.normalizer import normalize_task_report
from src.models.report import ReportStatus
report = normalize_task_report(self._llm_task(fail_count=1, warn_count=5))
assert report.status == ReportStatus.FAILED
# #endregion Test.ReportNormalizer.TestNormalizeTaskReportLlmValidation
# #region Test.ReportNormalizer.TestNormalizeTaskReportSourceRef [C:2] [TYPE Function]
# @BRIEF Test normalize_task_report source_ref when params is not a dict.
class TestNormalizeTaskReportSourceRef:
def test_params_not_dict_yields_none_source_ref(self):
from src.services.reports.normalizer import normalize_task_report
from src.core.task_manager.models import Task, TaskStatus
task = Task.model_construct(
id="task-1",
plugin_id="superset-migration",
status=TaskStatus.SUCCESS,
params=None, # bypasses Pydantic validation via model_construct
result=None,
started_at=datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC),
finished_at=datetime(2024, 1, 1, 12, 30, 0, tzinfo=UTC),
logs=[],
)
report = normalize_task_report(task)
assert report.source_ref is None
assert report.report_id == "task-1"
# #endregion Test.ReportNormalizer.TestNormalizeTaskReportSourceRef
# #endregion Test.ReportNormalizer