Skip to content
Snippets Groups Projects
Commit dc22fc8f authored by Carlos Natalino Da Silva's avatar Carlos Natalino Da Silva
Browse files

Fixing the optical attack manager tests.

parent 68ced8ae
No related branches found
No related tags found
2 merge requests!142Release TeraFlowSDN 2.1,!138Fixing the tests of the optical cybersecurity components
...@@ -25,96 +25,14 @@ from common.proto import dbscanserving_pb2 as dbscan ...@@ -25,96 +25,14 @@ from common.proto import dbscanserving_pb2 as dbscan
from common.proto.optical_attack_detector_pb2 import DetectionRequest from common.proto.optical_attack_detector_pb2 import DetectionRequest
from opticalattackdetector.client.OpticalAttackDetectorClient import \ from opticalattackdetector.client.OpticalAttackDetectorClient import \
OpticalAttackDetectorClient OpticalAttackDetectorClient
from opticalattackdetector.Config import GRPC_SERVICE_PORT
from opticalattackdetector.service.OpticalAttackDetectorService import \ from opticalattackdetector.service.OpticalAttackDetectorService import \
OpticalAttackDetectorService OpticalAttackDetectorService
# from .example_objects import CONTEXT_ID, CONTEXT_ID_2, SERVICE_DEV1_DEV2 # from .example_objects import CONTEXT_ID, CONTEXT_ID_2, SERVICE_DEV1_DEV2
port = 10000 + GRPC_SERVICE_PORT # avoid privileged ports
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG) LOGGER.setLevel(logging.DEBUG)
@pytest.fixture(scope="session") def test_detect_attack():
def optical_attack_detector_service():
with patch.dict(
os.environ,
{
"OPTICALATTACKDETECTORSERVICE_SERVICE_HOST": "127.0.0.1",
"OPTICALATTACKDETECTORSERVICE_SERVICE_PORT_GRPC": str(
1000
+ DEFAULT_SERVICE_GRPC_PORTS.get(
ServiceNameEnum.OPTICALATTACKDETECTOR.value
)
),
"OPTICALATTACKMITIGATORSERVICE_SERVICE_HOST": "127.0.0.1",
"OPTICALATTACKMITIGATORSERVICE_SERVICE_PORT_GRPC": str(
1000
+ DEFAULT_SERVICE_GRPC_PORTS.get(
ServiceNameEnum.OPTICALATTACKMITIGATOR.value
)
),
"DBSCANSERVINGSERVICE_SERVICE_HOST": "127.0.0.1",
"DBSCANSERVINGSERVICE_SERVICE_PORT_GRPC": str(
1000
+ DEFAULT_SERVICE_GRPC_PORTS.get(ServiceNameEnum.DBSCANSERVING.value)
),
},
clear=True,
):
_service = OpticalAttackDetectorService(
port=port,
max_workers=DEFAULT_GRPC_MAX_WORKERS,
grace_period=DEFAULT_GRPC_GRACE_PERIOD,
)
# mocker_context_client = mock.patch('opticalattackdetector.service.OpticalAttackDetectorServiceServicerImpl.context_client')
# mocker_context_client.start()
# mocker_influx_db = mock.patch('opticalattackdetector.service.OpticalAttackDetectorServiceServicerImpl.influxdb_client')
# mocker_influx_db.start()
_service.start()
yield _service
_service.stop()
# mocker_context_client.stop()
# mocker_influx_db.stop()
@pytest.fixture(scope="session")
def optical_attack_detector_client(optical_attack_detector_service):
with patch.dict(
os.environ,
{
"OPTICALATTACKDETECTORSERVICE_SERVICE_HOST": "127.0.0.1",
"OPTICALATTACKDETECTORSERVICE_SERVICE_PORT_GRPC": str(
1000
+ DEFAULT_SERVICE_GRPC_PORTS.get(
ServiceNameEnum.OPTICALATTACKDETECTOR.value
)
),
"OPTICALATTACKMITIGATORSERVICE_SERVICE_HOST": "127.0.0.1",
"OPTICALATTACKMITIGATORSERVICE_SERVICE_PORT_GRPC": str(
1000
+ DEFAULT_SERVICE_GRPC_PORTS.get(
ServiceNameEnum.OPTICALATTACKMITIGATOR.value
)
),
"DBSCANSERVINGSERVICE_SERVICE_HOST": "127.0.0.1",
"DBSCANSERVINGSERVICE_SERVICE_PORT_GRPC": str(
1000
+ DEFAULT_SERVICE_GRPC_PORTS.get(ServiceNameEnum.DBSCANSERVING.value)
),
},
clear=True,
):
_client = OpticalAttackDetectorClient()
yield _client
_client.close()
def test_detect_attack(
optical_attack_detector_client: OpticalAttackDetectorClient,
):
LOGGER.info("placeholder") LOGGER.info("placeholder")
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment