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

Island: Remove disused agent_event_handlers/utils.py 

Replaced by NodeUpdateFacade
This commit is contained in:
Mike Salvatore 2022-10-07 09:52:29 -04:00
parent e876682d84
commit b6a6295ae8
2 changed files with 0 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
monkey/monkey_island/cc/agent_event_handlers/utils.py
View File

@ -1,17 +0,0 @@
from ipaddress import IPv4Address, IPv4Interface
from monkey_island.cc.models import Machine
from monkey_island.cc.repository import IMachineRepository, UnknownRecordError
def get_or_create_target_machine(repository: IMachineRepository, target: IPv4Address):
try:
target_machines = repository.get_machines_by_ip(target)
return target_machines[0]
except UnknownRecordError:
machine = Machine(
id=repository.get_new_id(),
network_interfaces=[IPv4Interface(target)],
)
repository.upsert_machine(machine)
return machine

46
monkey/tests/unit_tests/monkey_island/cc/agent_event_handlers/test_utils.py
View File

@ -1,46 +0,0 @@
from ipaddress import IPv4Address, IPv4Interface
from unittest.mock import MagicMock
import pytest
from monkey_island.cc.agent_event_handlers.utils import get_or_create_target_machine
from monkey_island.cc.models import Machine
from monkey_island.cc.repository import IMachineRepository, UnknownRecordError
SEED_ID = 99
IP_ADDRESS = IPv4Address("10.10.10.99")
EXISTING_MACHINE = Machine(
id=1,
hardware_id=5,
network_interfaces=[IPv4Interface(IP_ADDRESS)],
)
EXPECTED_CREATED_MACHINE = Machine(
id=SEED_ID,
network_interfaces=[IPv4Interface(IP_ADDRESS)],
)
@pytest.fixture
def machine_repository() -> IMachineRepository:
machine_repository = MagicMock(spec=IMachineRepository)
machine_repository.get_new_id = MagicMock(return_value=SEED_ID)
return machine_repository
def test_return_existing_machine(machine_repository):
machine_repository.get_machines_by_ip = MagicMock(return_value=[EXISTING_MACHINE])
target_machine = get_or_create_target_machine(machine_repository, IP_ADDRESS)
assert target_machine == EXISTING_MACHINE
def test_create_new_machine(machine_repository):
machine_repository.get_machines_by_ip = MagicMock(side_effect=UnknownRecordError)
target_machine = get_or_create_target_machine(machine_repository, IP_ADDRESS)
assert target_machine == EXPECTED_CREATED_MACHINE
assert machine_repository.upsert_machine.called_once_with(target_machine)