p15693087
/
monkey
forked from p15670423/monkey
1
0
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

UT: Implement test_handle_scan_data__node_not_upserted_if_no_matching_agent

This commit is contained in:
Kekoa Kaaikala 2022-09-27 21:01:48 +00:00
parent ce8219aa6d
commit 18ca84a247
1 changed files with 19 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
monkey/tests/unit_tests/monkey_island/cc/agent_event_handlers/test_handle_scan_data.py
View File

@ -34,6 +34,11 @@ STORED_MACHINE = Machine(
hardware_id=9, hardware_id=9,
network_interfaces=[IPv4Interface("10.10.10.1/24")], network_interfaces=[IPv4Interface("10.10.10.1/24")],
) )
EVENT = PingScanEvent(
source=AGENT_ID,
target=IPv4Address("10.10.10.1"),
scan_data=PingScanData(True, OperatingSystem.LINUX),
)
@pytest.fixture @pytest.fixture
@ -67,15 +72,9 @@ def handler(agent_repository, machine_repository, node_repository) -> handle_sca
def test_handle_scan_data__upserts_machine( def test_handle_scan_data__upserts_machine(
handler: handle_scan_data, handler: handle_scan_data,
machine_repository: IMachineRepository, machine_repository: IMachineRepository,
node_repository: INodeRepository,
): ):
event = PingScanEvent(
source=AGENT_ID,
target=IPv4Address("10.10.10.1"),
scan_data=PingScanData(True, OperatingSystem.LINUX),
)
machine_repository.get_machine_by_id = MagicMock(return_value=STORED_MACHINE) machine_repository.get_machine_by_id = MagicMock(return_value=STORED_MACHINE)
handler(event) handler(EVENT)
expected_machine = STORED_MACHINE.copy() expected_machine = STORED_MACHINE.copy()
expected_machine.operating_system = OperatingSystem.LINUX expected_machine.operating_system = OperatingSystem.LINUX
@ -88,21 +87,26 @@ def test_handle_scan_data__upserts_node(
machine_repository: IMachineRepository, machine_repository: IMachineRepository,
node_repository: INodeRepository, node_repository: INodeRepository,
): ):
event = PingScanEvent(
source=AGENT_ID,
target=IPv4Address("10.10.10.1"),
scan_data=PingScanData(True, OperatingSystem.LINUX),
)
machine_repository.get_machine_by_id = MagicMock(return_value=STORED_MACHINE) machine_repository.get_machine_by_id = MagicMock(return_value=STORED_MACHINE)
handler(event) handler(EVENT)
assert node_repository.upsert_communication.called_with( assert node_repository.upsert_communication.called_with(
MACHINE.id, STORED_MACHINE.id, CommunicationType.SCANNED MACHINE.id, STORED_MACHINE.id, CommunicationType.SCANNED
) )
def test_handle_scan_data__node_not_upserted_if_no_matching_agent(handler: handle_scan_data): def test_handle_scan_data__node_not_upserted_if_no_matching_agent(
pass handler: handle_scan_data,
agent_repository: IAgentRepository,
machine_repository: IMachineRepository,
node_repository: INodeRepository,
):
agent_repository.get_agent_by_id = MagicMock(side_effect=UnknownRecordError)
machine_repository.get_machine_by_id = MagicMock(return_value=STORED_MACHINE)
handler(EVENT)
assert not node_repository.upsert_communication.called
def test_handle_scan_data__node_not_upserted_if_no_matching_machine(handler: handle_scan_data): def test_handle_scan_data__node_not_upserted_if_no_matching_machine(handler: handle_scan_data):