589 lines
24 KiB
Python
589 lines
24 KiB
Python
# #region Test.ReportsService [C:3] [TYPE Module] [SEMANTICS test,report,service,pagination,filter]
|
|
# @BRIEF Tests for services/reports/report_service.py — ReportsService.
|
|
# @RELATION BINDS_TO -> [report_service]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
# #region helper factories [C:1] [TYPE Function]
|
|
def _make_task(**overrides):
|
|
from src.core.task_manager.models import Task, TaskStatus
|
|
defaults = {
|
|
"id": "task-1",
|
|
"plugin_id": "superset-migration",
|
|
"status": TaskStatus.SUCCESS,
|
|
"params": {},
|
|
"result": {"summary": "Migration complete"},
|
|
"started_at": datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC),
|
|
"finished_at": datetime(2024, 1, 1, 12, 30, 0, tzinfo=UTC),
|
|
"logs": [],
|
|
}
|
|
defaults.update(overrides)
|
|
return Task(**defaults)
|
|
|
|
|
|
def _make_service(task_manager=None, clean_release_repo=None):
|
|
from src.services.reports.report_service import ReportsService
|
|
if task_manager is None:
|
|
task_manager = MagicMock()
|
|
task_manager.get_all_tasks.return_value = []
|
|
return ReportsService(task_manager, clean_release_repo)
|
|
# #endregion helper factories
|
|
|
|
|
|
# #region test_reports_service_init [C:2] [TYPE Function]
|
|
# @BRIEF Test ReportsService initialization.
|
|
class TestReportsServiceInit:
|
|
def test_stores_task_manager(self):
|
|
tm = MagicMock()
|
|
service = _make_service(task_manager=tm)
|
|
assert service.task_manager is tm
|
|
|
|
def test_stores_clean_release_repo(self):
|
|
crr = MagicMock()
|
|
service = _make_service(clean_release_repo=crr)
|
|
assert service.clean_release_repository is crr
|
|
|
|
def test_clean_release_repo_optional(self):
|
|
service = _make_service()
|
|
assert service.clean_release_repository is None
|
|
# #endregion test_reports_service_init
|
|
|
|
|
|
# #region test_load_normalized_reports [C:2] [TYPE Function]
|
|
# @BRIEF Test _load_normalized_reports.
|
|
class TestLoadNormalizedReports:
|
|
def test_returns_empty_list_for_no_tasks(self):
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = []
|
|
service = _make_service(task_manager=tm)
|
|
reports = service._load_normalized_reports()
|
|
assert reports == []
|
|
|
|
def test_normalizes_all_tasks(self):
|
|
tm = MagicMock()
|
|
task = _make_task()
|
|
tm.get_all_tasks.return_value = [task]
|
|
service = _make_service(task_manager=tm)
|
|
reports = service._load_normalized_reports()
|
|
assert len(reports) == 1
|
|
assert reports[0].report_id == task.id
|
|
# #endregion test_load_normalized_reports
|
|
|
|
|
|
# #region test_to_utc_datetime [C:2] [TYPE Function]
|
|
# @BRIEF Test _to_utc_datetime.
|
|
class TestToUtcDatetime:
|
|
def test_none_returns_none(self):
|
|
service = _make_service()
|
|
assert service._to_utc_datetime(None) is None
|
|
|
|
def test_naive_interpreted_as_utc(self):
|
|
service = _make_service()
|
|
dt = datetime(2024, 1, 1, 12, 0, 0)
|
|
result = service._to_utc_datetime(dt)
|
|
assert result.tzinfo is not None
|
|
assert result.tzinfo == UTC
|
|
|
|
def test_aware_converted_to_utc(self):
|
|
service = _make_service()
|
|
from zoneinfo import ZoneInfo
|
|
dt = datetime(2024, 1, 1, 15, 0, 0, tzinfo=ZoneInfo("Europe/Moscow"))
|
|
result = service._to_utc_datetime(dt)
|
|
assert result.tzinfo == UTC
|
|
assert result.hour == 12 # Moscow +3 -> UTC
|
|
# #endregion test_to_utc_datetime
|
|
|
|
|
|
# #region test_datetime_sort_key [C:2] [TYPE Function]
|
|
# @BRIEF Test _datetime_sort_key.
|
|
class TestDatetimeSortKey:
|
|
def test_returns_timestamp(self):
|
|
from src.models.report import TaskReport, ReportStatus, TaskType
|
|
service = _make_service()
|
|
report = MagicMock(spec=TaskReport)
|
|
report.updated_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC)
|
|
key = service._datetime_sort_key(report)
|
|
assert key == 1704110400.0
|
|
|
|
def test_returns_zero_for_none(self):
|
|
from src.models.report import TaskReport
|
|
service = _make_service()
|
|
report = MagicMock(spec=TaskReport)
|
|
report.updated_at = None
|
|
key = service._datetime_sort_key(report)
|
|
assert key == 0.0
|
|
# #endregion test_datetime_sort_key
|
|
|
|
|
|
# #region test_matches_query [C:2] [TYPE Function]
|
|
# @BRIEF Test _matches_query.
|
|
class TestMatchesQuery:
|
|
def test_no_filters_matches_all(self):
|
|
from src.models.report import ReportQuery
|
|
service = _make_service()
|
|
report = MagicMock()
|
|
query = ReportQuery()
|
|
assert service._matches_query(report, query) is True
|
|
|
|
def test_filters_by_task_type(self):
|
|
from src.models.report import ReportQuery, TaskType
|
|
service = _make_service()
|
|
report = MagicMock()
|
|
report.task_type = TaskType.MIGRATION
|
|
query = ReportQuery(task_types=[TaskType.MIGRATION])
|
|
assert service._matches_query(report, query) is True
|
|
query = ReportQuery(task_types=[TaskType.BACKUP])
|
|
assert service._matches_query(report, query) is False
|
|
|
|
def test_filters_by_status(self):
|
|
from src.models.report import ReportQuery, ReportStatus
|
|
service = _make_service()
|
|
report = MagicMock()
|
|
report.status = ReportStatus.SUCCESS
|
|
query = ReportQuery(statuses=[ReportStatus.SUCCESS])
|
|
assert service._matches_query(report, query) is True
|
|
query = ReportQuery(statuses=[ReportStatus.FAILED])
|
|
assert service._matches_query(report, query) is False
|
|
|
|
def test_filters_by_time_range(self):
|
|
from src.models.report import ReportQuery
|
|
service = _make_service()
|
|
report = MagicMock()
|
|
report.updated_at = datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC)
|
|
# Time from before -> match
|
|
query = ReportQuery(time_from=datetime(2024, 1, 1, tzinfo=UTC))
|
|
assert service._matches_query(report, query) is True
|
|
# Time from after -> no match
|
|
query = ReportQuery(time_from=datetime(2024, 12, 1, tzinfo=UTC))
|
|
assert service._matches_query(report, query) is False
|
|
# Time to before -> no match
|
|
query = ReportQuery(time_to=datetime(2024, 1, 1, tzinfo=UTC))
|
|
assert service._matches_query(report, query) is False
|
|
|
|
def test_filters_by_search(self):
|
|
from src.models.report import ReportQuery, ReportStatus, TaskType
|
|
service = _make_service()
|
|
report = MagicMock()
|
|
report.summary = "Migration completed"
|
|
report.task_type = TaskType.MIGRATION
|
|
report.status = ReportStatus.SUCCESS
|
|
query = ReportQuery(search="migration")
|
|
assert service._matches_query(report, query) is True
|
|
query = ReportQuery(search="backup")
|
|
assert service._matches_query(report, query) is False
|
|
# #endregion test_matches_query
|
|
|
|
|
|
# #region test_sort_reports [C:2] [TYPE Function]
|
|
# @BRIEF Test _sort_reports.
|
|
class TestSortReports:
|
|
def test_sort_by_updated_at_desc_default(self):
|
|
from src.models.report import ReportQuery
|
|
service = _make_service()
|
|
reports = [
|
|
MagicMock(updated_at=datetime(2024, 1, 1, tzinfo=UTC)),
|
|
MagicMock(updated_at=datetime(2024, 6, 15, tzinfo=UTC)),
|
|
MagicMock(updated_at=datetime(2024, 3, 1, tzinfo=UTC)),
|
|
]
|
|
query = ReportQuery()
|
|
sorted_reports = service._sort_reports(reports, query)
|
|
assert sorted_reports[0].updated_at > sorted_reports[-1].updated_at
|
|
|
|
def test_sort_by_status_asc(self):
|
|
from src.models.report import ReportQuery, ReportStatus
|
|
service = _make_service()
|
|
reports = [
|
|
MagicMock(status=ReportStatus.FAILED, updated_at=datetime(2024, 1, 1, tzinfo=UTC)),
|
|
MagicMock(status=ReportStatus.SUCCESS, updated_at=datetime(2024, 6, 15, tzinfo=UTC)),
|
|
]
|
|
query = ReportQuery(sort_by="status", sort_order="asc")
|
|
sorted_reports = service._sort_reports(reports, query)
|
|
assert sorted_reports[0].status.value < sorted_reports[1].status.value
|
|
|
|
def test_sort_by_task_type_desc(self):
|
|
from src.models.report import ReportQuery, TaskType
|
|
service = _make_service()
|
|
reports = [
|
|
MagicMock(task_type=TaskType.BACKUP, updated_at=datetime(2024, 1, 1, tzinfo=UTC)),
|
|
MagicMock(task_type=TaskType.MIGRATION, updated_at=datetime(2024, 6, 15, tzinfo=UTC)),
|
|
]
|
|
query = ReportQuery(sort_by="task_type", sort_order="desc")
|
|
sorted_reports = service._sort_reports(reports, query)
|
|
assert sorted_reports[0].task_type.value > sorted_reports[1].task_type.value
|
|
# #endregion test_sort_reports
|
|
|
|
|
|
# #region test_list_reports [C:2] [TYPE Function]
|
|
# @BRIEF Test list_reports with pagination.
|
|
class TestListReports:
|
|
def test_returns_empty_collection(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = []
|
|
service = _make_service(task_manager=tm)
|
|
result = service.list_reports(ReportQuery())
|
|
assert result.total == 0
|
|
assert result.items == []
|
|
assert result.has_next is False
|
|
|
|
def test_paginates_results(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tasks = [_make_task(id=f"task-{i}", plugin_id="test") for i in range(5)]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
query = ReportQuery(page=1, page_size=2)
|
|
result = service.list_reports(query)
|
|
assert result.total == 5
|
|
assert len(result.items) == 2
|
|
assert result.has_next is True
|
|
|
|
def test_last_page_no_next(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tasks = [_make_task(id=f"task-{i}", plugin_id="test") for i in range(3)]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
query = ReportQuery(page=2, page_size=2)
|
|
result = service.list_reports(query)
|
|
assert result.total == 3
|
|
assert len(result.items) == 1
|
|
assert result.has_next is False
|
|
|
|
def test_filters_applied(self):
|
|
from src.models.report import ReportQuery, TaskType
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
tasks = [
|
|
_make_task(id="task-backup", plugin_id="superset-backup"),
|
|
_make_task(id="task-migration", plugin_id="superset-migration"),
|
|
]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
query = ReportQuery(task_types=[TaskType.BACKUP])
|
|
result = service.list_reports(query)
|
|
assert result.total == 1
|
|
assert result.items[0].task_type == TaskType.BACKUP
|
|
|
|
def test_search_filter(self):
|
|
from src.models.report import ReportQuery
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
tasks = [
|
|
_make_task(id="t1", plugin_id="superset-migration", result={"summary": "Backup completed"}),
|
|
_make_task(id="t2", plugin_id="superset-backup", result={"summary": "Migration done"}),
|
|
]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
query = ReportQuery(search="backup")
|
|
result = service.list_reports(query)
|
|
# Both "Backup" and "backup" summaries match
|
|
assert result.total >= 1
|
|
# #endregion test_list_reports
|
|
|
|
|
|
# #region test_get_summary [C:2] [TYPE Function]
|
|
# @BRIEF Test get_summary aggregation.
|
|
# @TEST_EDGE: all statuses distributed into 5 buckets, empty tasks returns zeros, pending/awaiting_input counted separately.
|
|
class TestGetSummary:
|
|
def test_empty_tasks_returns_zeros(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = []
|
|
service = _make_service(task_manager=tm)
|
|
summary = service.get_summary()
|
|
assert summary.total_tasks == 0
|
|
assert summary.active_tasks == 0
|
|
assert summary.by_type == []
|
|
|
|
def test_counts_all_5_buckets_separately(self):
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
tasks = [
|
|
_make_task(id="t-pending", status=TaskStatus.PENDING),
|
|
_make_task(id="t-running", status=TaskStatus.RUNNING),
|
|
_make_task(id="t-awaiting", status=TaskStatus.AWAITING_INPUT, input_request={"message": "input needed"}),
|
|
_make_task(id="t-success", status=TaskStatus.SUCCESS),
|
|
_make_task(id="t-failed", status=TaskStatus.FAILED),
|
|
]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
summary = service.get_summary()
|
|
assert len(summary.by_type) == 1 # all migration
|
|
ts = summary.by_type[0]
|
|
assert ts.counts.pending == 1
|
|
assert ts.counts.running == 1
|
|
assert ts.counts.awaiting_input == 1
|
|
assert ts.counts.success == 1
|
|
assert ts.counts.failed == 1
|
|
assert ts.total == 5
|
|
assert summary.total_tasks == 5
|
|
assert summary.active_tasks == 3 # pending + running + awaiting_input
|
|
|
|
def test_multiple_types_grouped(self):
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
tasks = [
|
|
_make_task(id="t1", plugin_id="superset-backup", status=TaskStatus.SUCCESS),
|
|
_make_task(id="t2", plugin_id="superset-migration", status=TaskStatus.FAILED),
|
|
_make_task(id="t3", plugin_id="superset-backup", status=TaskStatus.PENDING),
|
|
]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
summary = service.get_summary()
|
|
assert len(summary.by_type) == 2
|
|
for ts in summary.by_type:
|
|
if "backup" in ts.task_type.value:
|
|
assert ts.total == 2 # SUCCESS + PENDING
|
|
assert ts.counts.success == 1
|
|
assert ts.counts.pending == 1
|
|
elif "migration" in ts.task_type.value:
|
|
assert ts.total == 1
|
|
assert ts.counts.failed == 1
|
|
|
|
def test_pending_not_zero(self):
|
|
"""Verify PENDING/AWAITING_INPUT populate separate buckets, not collapsed into running."""
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
tasks = [
|
|
_make_task(id="t-pending", status=TaskStatus.PENDING),
|
|
_make_task(id="t-running", status=TaskStatus.RUNNING),
|
|
_make_task(id="t-awaiting", status=TaskStatus.AWAITING_INPUT, input_request={"message": "input needed"}),
|
|
]
|
|
tm.get_all_tasks.return_value = tasks
|
|
service = _make_service(task_manager=tm)
|
|
summary = service.get_summary()
|
|
ts = summary.by_type[0]
|
|
assert ts.counts.pending == 1
|
|
assert ts.counts.awaiting_input == 1
|
|
assert ts.counts.running == 1 # separate from pending/awaiting
|
|
# #endregion test_get_summary
|
|
|
|
|
|
# #region test_rbac_filtering [C:3] [TYPE Function] [SEMANTICS test,rbac,task-status-center]
|
|
# @BRIEF Rejected-path regression tests for ReportsService RBAC task filtering.
|
|
# @TEST_EDGE viewer_sees_only_own_tasks; analyst_sees_own_and_system; admin_sees_all.
|
|
class TestRbacFiltering:
|
|
def _user(self, user_id: str, role_name: str, is_admin: bool = False):
|
|
role = MagicMock()
|
|
role.name = role_name
|
|
role.is_admin = is_admin
|
|
user = MagicMock()
|
|
user.id = user_id
|
|
user.roles = [role]
|
|
return user
|
|
|
|
def _task_set(self):
|
|
return [
|
|
_make_task(id="own", user_id="user-1"),
|
|
_make_task(id="other", user_id="user-2"),
|
|
_make_task(id="system", user_id=None),
|
|
]
|
|
|
|
def test_viewer_sees_only_own_tasks(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = self._task_set()
|
|
service = _make_service(task_manager=tm)
|
|
|
|
result = service.list_reports(ReportQuery(), current_user=self._user("user-1", "viewer"))
|
|
|
|
assert {item.task_id for item in result.items} == {"own"}
|
|
|
|
def test_analyst_sees_own_and_system_tasks(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = self._task_set()
|
|
service = _make_service(task_manager=tm)
|
|
|
|
result = service.list_reports(ReportQuery(), current_user=self._user("user-1", "analyst"))
|
|
|
|
assert {item.task_id for item in result.items} == {"own", "system"}
|
|
|
|
def test_admin_sees_all_tasks(self):
|
|
from src.models.report import ReportQuery
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = self._task_set()
|
|
service = _make_service(task_manager=tm)
|
|
|
|
result = service.list_reports(ReportQuery(), current_user=self._user("user-1", "Admin", True))
|
|
|
|
assert {item.task_id for item in result.items} == {"own", "other", "system"}
|
|
|
|
def test_summary_uses_same_rbac_filter(self):
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = self._task_set()
|
|
service = _make_service(task_manager=tm)
|
|
|
|
viewer_summary = service.get_summary(current_user=self._user("user-1", "viewer"))
|
|
analyst_summary = service.get_summary(current_user=self._user("user-1", "analyst"))
|
|
|
|
assert viewer_summary.total_tasks == 1
|
|
assert analyst_summary.total_tasks == 2
|
|
# #endregion test_rbac_filtering
|
|
|
|
|
|
# #region test_get_report_detail [C:2] [TYPE Function]
|
|
# @BRIEF Test get_report_detail.
|
|
class TestGetReportDetail:
|
|
def test_returns_none_for_missing(self):
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = []
|
|
service = _make_service(task_manager=tm)
|
|
result = service.get_report_detail("nonexistent")
|
|
assert result is None
|
|
|
|
def test_returns_detail_for_existing(self):
|
|
tm = MagicMock()
|
|
task = _make_task(id="task-1")
|
|
tm.get_all_tasks.return_value = [task]
|
|
service = _make_service(task_manager=tm)
|
|
result = service.get_report_detail("task-1")
|
|
assert result is not None
|
|
assert result.report.report_id == "task-1"
|
|
assert len(result.timeline) == 2 # started + updated
|
|
|
|
def test_recovery_actions_for_failed(self):
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
task = _make_task(id="task-1", status=TaskStatus.FAILED, result={"error": {"message": "Failed"}})
|
|
tm.get_all_tasks.return_value = [task]
|
|
service = _make_service(task_manager=tm)
|
|
result = service.get_report_detail("task-1")
|
|
assert result is not None
|
|
assert len(result.next_actions) > 0
|
|
|
|
def test_default_next_actions_for_failed(self):
|
|
from src.core.task_manager.models import TaskStatus
|
|
tm = MagicMock()
|
|
task = _make_task(id="task-1", status=TaskStatus.FAILED, result=None)
|
|
tm.get_all_tasks.return_value = [task]
|
|
service = _make_service(task_manager=tm)
|
|
result = service.get_report_detail("task-1")
|
|
assert result is not None
|
|
assert "Review task diagnostics" in result.next_actions
|
|
|
|
def test_clean_release_enriches_detail(self):
|
|
from src.core.task_manager.models import TaskStatus
|
|
from src.models.report import TaskType
|
|
# Create a task that maps to CLEAN_RELEASE type
|
|
tm = MagicMock()
|
|
task = _make_task(id="cr-1", plugin_id="clean-release-compliance",
|
|
result={"run_id": "run-1"})
|
|
tm.get_all_tasks.return_value = [task]
|
|
|
|
crr = MagicMock()
|
|
mock_run = MagicMock()
|
|
mock_run.id = "run-1"
|
|
mock_run.candidate_id = "cand-1"
|
|
mock_run.status = "completed"
|
|
mock_run.final_status = "approved"
|
|
mock_run.requested_by = "admin"
|
|
crr.get_check_run.return_value = mock_run
|
|
crr.reports = {}
|
|
|
|
service = _make_service(task_manager=tm, clean_release_repo=crr)
|
|
result = service.get_report_detail("cr-1")
|
|
assert result is not None
|
|
assert "clean_release_run" in result.diagnostics
|
|
|
|
def test_clean_release_report_linked(self):
|
|
tm = MagicMock()
|
|
task = _make_task(id="cr-2", plugin_id="clean-release-compliance",
|
|
result={"run_id": "run-2"})
|
|
tm.get_all_tasks.return_value = [task]
|
|
|
|
crr = MagicMock()
|
|
mock_run = MagicMock()
|
|
mock_run.id = "run-2"
|
|
mock_run.candidate_id = "cand-2"
|
|
mock_run.status = "completed"
|
|
mock_run.final_status = "approved"
|
|
mock_run.requested_by = "admin"
|
|
crr.get_check_run.return_value = mock_run
|
|
|
|
mock_report = MagicMock()
|
|
mock_report.id = "report-1"
|
|
mock_report.final_status = "compliant"
|
|
crr.reports = {"rep1": mock_report}
|
|
|
|
# Make the iteration find the report by run_id
|
|
mock_report.run_id = "run-2"
|
|
|
|
service = _make_service(task_manager=tm, clean_release_repo=crr)
|
|
result = service.get_report_detail("cr-2")
|
|
assert result is not None
|
|
assert "clean_release_report" in result.diagnostics
|
|
|
|
def test_diagnostics_not_provided(self):
|
|
tm = MagicMock()
|
|
task = _make_task(id="task-1", result=None)
|
|
tm.get_all_tasks.return_value = [task]
|
|
service = _make_service(task_manager=tm)
|
|
result = service.get_report_detail("task-1")
|
|
assert result is not None
|
|
assert result.diagnostics is not None
|
|
|
|
def test_diagnostics_empty_details_uses_fallback(self):
|
|
"""When details is empty/falsy, uses 'Not provided' fallback (line 247)."""
|
|
from src.models.report import TaskReport, ReportStatus, TaskType
|
|
from unittest.mock import PropertyMock
|
|
from datetime import UTC, datetime
|
|
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = []
|
|
service = _make_service(task_manager=tm)
|
|
|
|
# Mock _load_normalized_reports to return a report with empty details
|
|
mock_report = MagicMock(spec=TaskReport)
|
|
mock_report.report_id = "empty-details-task"
|
|
mock_report.task_type = TaskType.MIGRATION
|
|
mock_report.status = ReportStatus.SUCCESS
|
|
mock_report.updated_at = datetime(2024, 1, 1, tzinfo=UTC)
|
|
mock_report.started_at = None
|
|
mock_report.error_context = None
|
|
type(mock_report).details = PropertyMock(return_value={})
|
|
|
|
with patch.object(service, "_load_normalized_reports", return_value=[mock_report]):
|
|
result = service.get_report_detail("empty-details-task")
|
|
assert result is not None
|
|
assert result.diagnostics == {"note": "Not provided"}
|
|
|
|
def test_default_next_actions_for_failed_without_error_context_actions(self):
|
|
"""When error_context exists but next_actions is empty, uses default (line 290)."""
|
|
from src.models.report import TaskReport, ReportStatus, TaskType, ErrorContext
|
|
from datetime import UTC, datetime
|
|
|
|
tm = MagicMock()
|
|
tm.get_all_tasks.return_value = []
|
|
service = _make_service(task_manager=tm)
|
|
|
|
mock_report = MagicMock(spec=TaskReport)
|
|
mock_report.report_id = "failed-no-actions"
|
|
mock_report.task_type = TaskType.MIGRATION
|
|
mock_report.status = ReportStatus.FAILED
|
|
mock_report.updated_at = datetime(2024, 1, 1, tzinfo=UTC)
|
|
mock_report.started_at = None
|
|
mock_report.details = {"result": {}}
|
|
mock_report.error_context = ErrorContext(
|
|
message="test error",
|
|
code=None,
|
|
next_actions=[],
|
|
)
|
|
|
|
with patch.object(service, "_load_normalized_reports", return_value=[mock_report]):
|
|
result = service.get_report_detail("failed-no-actions")
|
|
assert result is not None
|
|
assert "Review diagnostics" in result.next_actions
|
|
assert "Retry task if applicable" in result.next_actions
|
|
# #endregion test_get_report_detail
|
|
# #endregion Test.ReportsService
|