63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
|
"""FakeAdapter M2 tests — list_callbacks shape, create_task monotonicity."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from backend.app.services.c2.fake import FakeAdapter
|
||
|
|
|
||
|
|
|
||
|
|
class TestFakeAdapterListCallbacks:
|
||
|
|
def test_returns_three_callbacks(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
callbacks = adapter.list_callbacks()
|
||
|
|
assert len(callbacks) == 3
|
||
|
|
|
||
|
|
def test_all_active(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
for cb in adapter.list_callbacks():
|
||
|
|
assert cb.active is True
|
||
|
|
|
||
|
|
def test_display_ids_are_1_2_3(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
ids = [cb.display_id for cb in adapter.list_callbacks()]
|
||
|
|
assert ids == [1, 2, 3]
|
||
|
|
|
||
|
|
def test_pinned_last_checkin_format(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
for cb in adapter.list_callbacks():
|
||
|
|
assert cb.last_checkin.startswith("2026-06-10")
|
||
|
|
|
||
|
|
def test_callbacks_have_host_user_domain(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
for cb in adapter.list_callbacks():
|
||
|
|
assert cb.host
|
||
|
|
assert cb.user
|
||
|
|
assert cb.domain
|
||
|
|
|
||
|
|
|
||
|
|
class TestFakeAdapterCreateTask:
|
||
|
|
def test_returns_monotonic_ids_from_1000(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
id1 = adapter.create_task(1, "whoami")
|
||
|
|
id2 = adapter.create_task(1, "ipconfig")
|
||
|
|
assert id1 == 1000
|
||
|
|
assert id2 == 1001
|
||
|
|
|
||
|
|
def test_separate_instances_start_at_1000_independently(self):
|
||
|
|
a1 = FakeAdapter()
|
||
|
|
a2 = FakeAdapter()
|
||
|
|
assert a1.create_task(1, "cmd") == 1000
|
||
|
|
assert a2.create_task(1, "cmd") == 1000
|
||
|
|
|
||
|
|
def test_stores_command_and_callback(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
tid = adapter.create_task(callback_display_id=2, command="ls", params="-la")
|
||
|
|
task = adapter._tasks[tid]
|
||
|
|
assert task["command"] == "ls"
|
||
|
|
assert task["params"] == "-la"
|
||
|
|
assert task["callback_display_id"] == 2
|
||
|
|
|
||
|
|
def test_initial_status_submitted(self):
|
||
|
|
adapter = FakeAdapter()
|
||
|
|
tid = adapter.create_task(1, "hostname")
|
||
|
|
assert adapter._tasks[tid]["status"] == "submitted"
|
||
|
|
assert adapter._tasks[tid]["completed"] is False
|