8f23f59601
Command source decision: extended C2TaskStatus with command: str | None (default None). Added command_name to _GET_TASK_QUERY so get_task() returns command in a single round-trip — no separate history fetch needed on import. 4-line change, zero cascading test impact. adapter.py: - C2TaskStatus: add command: str | None = None field - C2HistoricalTask: new dataclass (display_id, command, params, status, completed, timestamp) for history rows - C2TaskPage.items: typed as list[C2HistoricalTask] (was list[dict]) mythic.py: - _GET_TASK_QUERY: add command_name field - _LIST_CALLBACK_TASKS_QUERY: new query (order_by id desc, limit/offset) - _COUNT_CALLBACK_TASKS_QUERY: new aggregate query for total - get_task(): surfaces command_name as status.command - list_callback_tasks(): two _post() calls (tasks + count), allow_redirects=False fake.py: - _FAKE_HISTORY: frozen deterministic history (cb1=12, cb2=0, cb3=5 tasks) - list_callback_tasks(): serves from _FAKE_HISTORY, pagination applied - get_task(): returns command from _tasks dict api/c2.py: - GET /api/engagements/<eid>/c2/callbacks/<cid>/history: page+page_size defaults 1/25, cap 100, reject <1, 502 on adapter error - POST /api/simulations/<sid>/c2/import: idempotent per (sim,mythic_id) pair, source=import, completed tasks get output+mapping_applied, incomplete tasks stored for poll-on-read pickup, auto-transition pending→in_progress 60 new tests (456 total); pytest/ruff/mypy all green Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
76 lines
3.0 KiB
Python
76 lines
3.0 KiB
Python
"""FakeAdapter M4 tests — list_callback_tasks pagination."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.c2.adapter import C2HistoricalTask
|
|
from backend.app.services.c2.fake import FakeAdapter
|
|
|
|
|
|
@pytest.fixture()
|
|
def adapter() -> FakeAdapter:
|
|
return FakeAdapter()
|
|
|
|
|
|
class TestFakeAdapterListCallbackTasks:
|
|
def test_callback_1_returns_12_total(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=1, page_size=25)
|
|
assert page.total == 12
|
|
|
|
def test_callback_2_returns_0_tasks(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=2, page=1, page_size=25)
|
|
assert page.total == 0
|
|
assert page.items == []
|
|
|
|
def test_callback_3_returns_5_tasks(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=3, page=1, page_size=25)
|
|
assert page.total == 5
|
|
assert len(page.items) == 5
|
|
|
|
def test_items_are_c2_historical_task_instances(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=1, page_size=5)
|
|
for item in page.items:
|
|
assert isinstance(item, C2HistoricalTask)
|
|
|
|
def test_pagination_page1(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=1, page_size=5)
|
|
assert len(page.items) == 5
|
|
assert page.page == 1
|
|
assert page.page_size == 5
|
|
|
|
def test_pagination_page2(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=2, page_size=5)
|
|
assert len(page.items) == 5
|
|
assert page.page == 2
|
|
|
|
def test_pagination_last_page_partial(self, adapter):
|
|
# 12 tasks, page_size=5 → page 3 has 2 items.
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=3, page_size=5)
|
|
assert len(page.items) == 2
|
|
assert page.total == 12
|
|
|
|
def test_pagination_beyond_range_returns_empty(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=99, page_size=25)
|
|
assert len(page.items) == 0
|
|
assert page.total == 12
|
|
|
|
def test_history_is_deterministic_across_instances(self):
|
|
a1 = FakeAdapter()
|
|
a2 = FakeAdapter()
|
|
p1 = a1.list_callback_tasks(callback_display_id=1, page=1, page_size=25)
|
|
p2 = a2.list_callback_tasks(callback_display_id=1, page=1, page_size=25)
|
|
assert [t.display_id for t in p1.items] == [t.display_id for t in p2.items]
|
|
|
|
def test_completed_and_submitted_mix(self, adapter):
|
|
"""Callback 1 has alternating completed/submitted tasks (even=completed)."""
|
|
page = adapter.list_callback_tasks(callback_display_id=1, page=1, page_size=12)
|
|
completed = [t for t in page.items if t.completed]
|
|
submitted = [t for t in page.items if not t.completed]
|
|
assert len(completed) == 6
|
|
assert len(submitted) == 6
|
|
|
|
def test_unknown_callback_returns_empty(self, adapter):
|
|
page = adapter.list_callback_tasks(callback_display_id=999, page=1, page_size=25)
|
|
assert page.total == 0
|
|
assert page.items == []
|