Initial fix of the attack detector tests. (eab4d7bc) · Commits · TFS / controller

src/opticalattackdetector/.gitlab-ci.yml

+19 −1

Original line number	Diff line number	Diff line
		@@ -45,13 +45,30 @@ unit_test opticalattackdetector:
		stage: unit_test
		needs:
		- build opticalattackdetector
		- build dbscanserving
		- unit_test dbscanserving
		before_script:
		- docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY
		- if docker network list \| grep teraflowbridge; then echo "teraflowbridge is already created"; else docker network create -d bridge teraflowbridge; fi
		- if docker container ls \| grep $IMAGE_NAME; then docker rm -f $IMAGE_NAME; else echo "$IMAGE_NAME image is not in the system"; fi
		- if docker container ls \| grep redis; then docker rm -f redis; else echo "redis image is not in the system"; fi
		- if docker container ls \| grep dbscanserving; then docker rm -f dbscanserving; else echo "dbscanserving image is not in the system"; fi
		script:
		- export REDIS_PASSWORD=`uuidgen`
		- docker pull "redis:7.0-alpine"
		- docker run --name redis -d --network=teraflowbridge -p 6379:6379 -e REDIS_PASSWORD=${REDIS_PASSWORD} --rm redis:7.0-alpine "redis-server --requirepass $(REDIS_PASSWORD)"
		- REDIS_ADDRESS=$(docker inspect redis --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}")
		- docker pull "$CI_REGISTRY_IMAGE/dbscanserving:$IMAGE_TAG"
		- docker run --name dbscanserving -d -p 10008:10008 --network=teraflowbridge --rm $CI_REGISTRY_IMAGE/dbscanserving:$IMAGE_TAG
		- DBSCANSERVING_ADDRESS=$(docker inspect dbscanserving --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}")
		- docker pull "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG"
		- docker run --name $IMAGE_NAME -d -p 10006:10006 --env-file "$PWD/src/$IMAGE_NAME/.env" -v "$PWD/src/$IMAGE_NAME/tests:/home/${IMAGE_NAME}/results" --network=teraflowbridge --rm $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG
		- >
		docker run --name $IMAGE_NAME -d -p 10006:10006
		-v "$PWD/src/$IMAGE_NAME/tests:/home/${IMAGE_NAME}/results"
		-e REDIS_PASSWORD=${REDIS_PASSWORD}
		-e DBSCANSERVINGSERVICE_SERVICE_HOST=${DBSCANSERVING_ADDRESS}
		-e CACHINGSERVICE_SERVICE_HOST=${REDIS_ADDRESS}
		--network=teraflowbridge --rm $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG
		- sleep 5
		- docker ps -a
		- docker logs $IMAGE_NAME
		@@ -59,6 +76,7 @@ unit_test opticalattackdetector:
		coverage: '/TOTAL\s+\d+\s+\d+\s+(\d+%)/'
		after_script:
		- docker rm -f $IMAGE_NAME
		- docker rm -f redis
		- docker network rm teraflowbridge
		rules:
		- if: '$CI_PIPELINE_SOURCE == "merge_request_event" && ($CI_MERGE_REQUEST_TARGET_BRANCH_NAME == "develop" \|\| $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == $CI_DEFAULT_BRANCH)'

src/opticalattackdetector/tests/test_unitary.py

+7 −35

Original line number	Diff line number	Diff line
		@@ -17,42 +17,24 @@ import os
		from unittest.mock import patch

		import pytest
		from common.Constants import (DEFAULT_GRPC_GRACE_PERIOD,
		DEFAULT_GRPC_MAX_WORKERS)

		from common.proto import dbscanserving_pb2 as dbscan
		from common.proto.optical_attack_detector_pb2 import DetectionRequest

		from opticalattackdetector.client.OpticalAttackDetectorClient import \
		OpticalAttackDetectorClient
		from opticalattackdetector.Config import GRPC_SERVICE_PORT
		from opticalattackdetector.service.OpticalAttackDetectorService import \
		OpticalAttackDetectorService

		# 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.setLevel(logging.DEBUG)


		@pytest.fixture(scope="session")
		def optical_attack_detector_service():
		with patch.dict(
		os.environ,
		{
		"OPTICALATTACKDETECTORSERVICE_SERVICE_HOST": "127.0.0.1",
		"OPTICALATTACKDETECTORSERVICE_SERVICE_PORT_GRPC": str(port),
		"DBSCANSERVINGSERVICE_SERVICE_HOST": "127.0.0.1",
		"DBSCANSERVINGSERVICE_SERVICE_PORT_GRPC": str(port),
		},
		clear=True,
		):
		_service = OpticalAttackDetectorService(
		port=port,
		max_workers=DEFAULT_GRPC_MAX_WORKERS,
		grace_period=DEFAULT_GRPC_GRACE_PERIOD,
		)
		_service = OpticalAttackDetectorService()
		_service.start()
		yield _service
		_service.stop()
		@@ -60,16 +42,6 @@ def optical_attack_detector_service():

		@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(port),
		"DBSCANSERVINGSERVICE_SERVICE_HOST": "127.0.0.1",
		"DBSCANSERVINGSERVICE_SERVICE_PORT_GRPC": str(port),
		},
		clear=True,
		):
		_client = OpticalAttackDetectorClient()
		yield _client
		_client.close()