Context: (b29a7438) · Commits · TFS / controller

scripts/run_tests_locally-context.sh

+1 −3

Original line number	Diff line number	Diff line
		@@ -60,9 +60,7 @@ export NATS_URI="nats://tfs:tfs123@172.254.254.11:4222"
		export PYTHONPATH=/home/tfs/tfs-ctrl/src
		# helpful pytest flags: --log-level=INFO -o log_cli=true --verbose --maxfail=1 --durations=0
		coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose --maxfail=1 \
		context/tests/test_context.py \
		context/tests/test_topology.py
		#context/tests/test_*.py
		context/tests/test_*.py

		echo
		echo "Coverage report:"

src/context/service/ContextServiceServicerImpl.py

+11 −10

Original line number	Diff line number	Diff line
		@@ -130,28 +130,29 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer

		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def ListDeviceIds(self, request : Empty, context : grpc.ServicerContext) -> DeviceIdList:
		return device_list_ids(self.db_engine)
		return DeviceIdList(device_ids=device_list_ids(self.db_engine))

		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def ListDevices(self, request : Empty, context : grpc.ServicerContext) -> DeviceList:
		return device_list_objs(self.db_engine)
		return DeviceList(devices=device_list_objs(self.db_engine))

		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def GetDevice(self, request : ContextId, context : grpc.ServicerContext) -> Device:
		return device_get(self.db_engine, request)
		return Device(**device_get(self.db_engine, request))

		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def SetDevice(self, request : Device, context : grpc.ServicerContext) -> DeviceId:
		device_id,updated = device_set(self.db_engine, request) # pylint: disable=unused-variable
		#event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE
		#notify_event(self.messagebroker, TOPIC_DEVICE, event_type, {'device_id': device_id})
		return device_id
		device_id,updated = device_set(self.db_engine, request)
		event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE
		notify_event(self.messagebroker, TOPIC_DEVICE, event_type, {'device_id': device_id})
		return DeviceId(**device_id)

		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
		def RemoveDevice(self, request : DeviceId, context : grpc.ServicerContext) -> Empty:
		deleted = device_delete(self.db_engine, request) # pylint: disable=unused-variable
		#if deleted:
		# notify_event(self.messagebroker, TOPIC_DEVICE, EventTypeEnum.EVENTTYPE_REMOVE, {'device_id': request})
		device_id,deleted = device_delete(self.db_engine, request)
		if deleted:
		event_type = EventTypeEnum.EVENTTYPE_REMOVE
		notify_event(self.messagebroker, TOPIC_DEVICE, event_type, {'device_id': device_id})
		return Empty()

		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

src/context/service/database/ConfigRule.py

+23 −4

Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import datetime, logging
		from sqlalchemy import delete
		from sqlalchemy.dialects.postgresql import insert
		from sqlalchemy.orm import Session
		@@ -22,8 +23,10 @@ from .models.enums.ConfigAction import grpc_to_enum__config_action
		from .models.ConfigRuleModel import ConfigRuleKindEnum, ConfigRuleModel
		from .uuids._Builder import get_uuid_random

		LOGGER = logging.getLogger(__name__)

		def compose_config_rules_data(
		config_rules : List[ConfigRule],
		config_rules : List[ConfigRule], now : datetime.datetime,
		device_uuid : Optional[str] = None, service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None
		) -> List[Dict]:
		dict_config_rules : List[Dict] = list()
		@@ -36,6 +39,8 @@ def compose_config_rules_data(
		'kind' : ConfigRuleKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member
		'action' : grpc_to_enum__config_action(config_rule.action),
		'data' : grpc_message_to_json_string(getattr(config_rule, str_kind, {})),
		'created_at' : now,
		'updated_at' : now,
		}
		if device_uuid is not None: dict_config_rule['device_uuid' ] = device_uuid
		if service_uuid is not None: dict_config_rule['service_uuid'] = service_uuid
		@@ -45,16 +50,30 @@ def compose_config_rules_data(

		def upsert_config_rules(
		session : Session, config_rules : List[Dict],
		device_uuid : Optional[str] = None, service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None
		) -> None:
		device_uuid : Optional[str] = None, service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None,
		) -> bool:
		# TODO: do not delete all rules; just add-remove as needed
		stmt = delete(ConfigRuleModel)
		if device_uuid is not None: stmt = stmt.where(ConfigRuleModel.device_uuid == device_uuid )
		if service_uuid is not None: stmt = stmt.where(ConfigRuleModel.service_uuid == service_uuid)
		if slice_uuid is not None: stmt = stmt.where(ConfigRuleModel.slice_uuid == slice_uuid )
		session.execute(stmt)

		updated = False
		if len(config_rules) > 0:
		session.execute(insert(ConfigRuleModel).values(config_rules))
		stmt = insert(ConfigRuleModel).values(config_rules)
		#stmt = stmt.on_conflict_do_update(
		# index_elements=[ConfigRuleModel.configrule_uuid],
		# set_=dict(
		# updated_at = stmt.excluded.updated_at,
		# )
		#)
		stmt = stmt.returning(ConfigRuleModel.created_at, ConfigRuleModel.updated_at)
		config_rule_updates = session.execute(stmt).fetchall()
		LOGGER.warning('config_rule_updates = {:s}'.format(str(config_rule_updates)))
		# TODO: updated = ...

		return updated

		#Union_SpecificConfigRule = Union[
		# ConfigRuleCustomModel, ConfigRuleAclModel

src/context/service/database/Device.py

+36 −21

Original line number	Diff line number	Diff line
		@@ -12,15 +12,15 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import datetime, logging
		from sqlalchemy.dialects.postgresql import insert
		from sqlalchemy.engine import Engine
		from sqlalchemy.orm import Session, sessionmaker
		from sqlalchemy_cockroachdb import run_transaction
		from typing import Dict, List, Optional, Set, Tuple
		from common.proto.context_pb2 import Device, DeviceId, DeviceIdList, DeviceList
		from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException
		from common.proto.context_pb2 import Device, DeviceId
		from common.tools.object_factory.Device import json_device_id
		from context.service.database.ConfigRule import compose_config_rules_data, upsert_config_rules
		from .models.DeviceModel import DeviceModel
		from .models.EndPointModel import EndPointModel
		from .models.TopologyModel import TopologyDeviceModel
		@@ -29,22 +29,23 @@ from .models.enums.DeviceOperationalStatus import grpc_to_enum__device_operation
		from .models.enums.KpiSampleType import grpc_to_enum__kpi_sample_type
		from .uuids.Device import device_get_uuid
		from .uuids.EndPoint import endpoint_get_uuid
		from .ConfigRule import compose_config_rules_data, upsert_config_rules

		def device_list_ids(db_engine : Engine) -> DeviceIdList:
		LOGGER = logging.getLogger(__name__)

		def device_list_ids(db_engine : Engine) -> List[Dict]:
		def callback(session : Session) -> List[Dict]:
		obj_list : List[DeviceModel] = session.query(DeviceModel).all()
		#.options(selectinload(DeviceModel.topology)).filter_by(context_uuid=context_uuid).one_or_none()
		return [obj.dump_id() for obj in obj_list]
		return DeviceIdList(device_ids=run_transaction(sessionmaker(bind=db_engine), callback))
		return run_transaction(sessionmaker(bind=db_engine), callback)

		def device_list_objs(db_engine : Engine) -> DeviceList:
		def device_list_objs(db_engine : Engine) -> List[Dict]:
		def callback(session : Session) -> List[Dict]:
		obj_list : List[DeviceModel] = session.query(DeviceModel).all()
		#.options(selectinload(DeviceModel.topology)).filter_by(context_uuid=context_uuid).one_or_none()
		return [obj.dump() for obj in obj_list]
		return DeviceList(devices=run_transaction(sessionmaker(bind=db_engine), callback))
		return run_transaction(sessionmaker(bind=db_engine), callback)

		def device_get(db_engine : Engine, request : DeviceId) -> Device:
		def device_get(db_engine : Engine, request : DeviceId) -> Dict:
		device_uuid = device_get_uuid(request, allow_random=False)
		def callback(session : Session) -> Optional[Dict]:
		obj : Optional[DeviceModel] = session.query(DeviceModel).filter_by(device_uuid=device_uuid).one_or_none()
		@@ -55,9 +56,9 @@ def device_get(db_engine : Engine, request : DeviceId) -> Device:
		raise NotFoundException('Device', raw_device_uuid, extra_details=[
		'device_uuid generated was: {:s}'.format(device_uuid)
		])
		return Device(**obj)
		return obj

		def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		def device_set(db_engine : Engine, request : Device) -> Tuple[Dict, bool]:
		raw_device_uuid = request.device_id.device_uuid.uuid
		raw_device_name = request.name
		device_name = raw_device_uuid if len(raw_device_name) == 0 else raw_device_name
		@@ -67,6 +68,8 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		oper_status = grpc_to_enum__device_operational_status(request.device_operational_status)
		device_drivers = [grpc_to_enum__device_driver(d) for d in request.device_drivers]

		now = datetime.datetime.utcnow()

		topology_uuids : Set[str] = set()
		related_topologies : List[Dict] = list()
		endpoints_data : List[Dict] = list()
		@@ -94,6 +97,8 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		'name' : endpoint_name,
		'endpoint_type' : endpoint.endpoint_type,
		'kpi_sample_types': kpi_sample_types,
		'created_at' : now,
		'updated_at' : now,
		})

		if endpoint_topology_uuid not in topology_uuids:
		@@ -103,7 +108,7 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		})
		topology_uuids.add(endpoint_topology_uuid)

		config_rules = compose_config_rules_data(request.device_config.config_rules, device_uuid=device_uuid)
		config_rules = compose_config_rules_data(request.device_config.config_rules, now, device_uuid=device_uuid)

		device_data = [{
		'device_uuid' : device_uuid,
		@@ -111,9 +116,11 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		'device_type' : device_type,
		'device_operational_status': oper_status,
		'device_drivers' : device_drivers,
		'created_at' : now,
		'updated_at' : now,
		}]

		def callback(session : Session) -> None:
		def callback(session : Session) -> bool:
		stmt = insert(DeviceModel).values(device_data)
		stmt = stmt.on_conflict_do_update(
		index_elements=[DeviceModel.device_uuid],
		@@ -122,9 +129,12 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		device_type = stmt.excluded.device_type,
		device_operational_status = stmt.excluded.device_operational_status,
		device_drivers = stmt.excluded.device_drivers,
		updated_at = stmt.excluded.updated_at,
		)
		)
		session.execute(stmt)
		stmt = stmt.returning(DeviceModel.created_at, DeviceModel.updated_at)
		created_at,updated_at = session.execute(stmt).fetchone()
		updated = updated_at > created_at

		stmt = insert(EndPointModel).values(endpoints_data)
		stmt = stmt.on_conflict_do_update(
		@@ -133,23 +143,28 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[DeviceId, bool]:
		name = stmt.excluded.name,
		endpoint_type = stmt.excluded.endpoint_type,
		kpi_sample_types = stmt.excluded.kpi_sample_types,
		updated_at = stmt.excluded.updated_at,
		)
		)
		session.execute(stmt)
		stmt = stmt.returning(EndPointModel.created_at, EndPointModel.updated_at)
		endpoint_updates = session.execute(stmt).fetchall()
		LOGGER.warning('endpoint_updates = {:s}'.format(str(endpoint_updates)))

		session.execute(insert(TopologyDeviceModel).values(related_topologies).on_conflict_do_nothing(
		index_elements=[TopologyDeviceModel.topology_uuid, TopologyDeviceModel.device_uuid]
		))

		upsert_config_rules(session, config_rules, device_uuid=device_uuid)
		configrules_updated = upsert_config_rules(session, config_rules, device_uuid=device_uuid)

		return updated

		run_transaction(sessionmaker(bind=db_engine), callback)
		updated = False # TODO: improve and check if created/updated
		return DeviceId(**json_device_id(device_uuid)),updated
		updated = run_transaction(sessionmaker(bind=db_engine), callback)
		return json_device_id(device_uuid),updated

		def device_delete(db_engine : Engine, request : DeviceId) -> bool:
		def device_delete(db_engine : Engine, request : DeviceId) -> Tuple[Dict, bool]:
		device_uuid = device_get_uuid(request, allow_random=False)
		def callback(session : Session) -> bool:
		num_deleted = session.query(DeviceModel).filter_by(device_uuid=device_uuid).delete()
		return num_deleted > 0
		return run_transaction(sessionmaker(bind=db_engine), callback)
		deleted = run_transaction(sessionmaker(bind=db_engine), callback)
		return json_device_id(device_uuid),deleted

src/context/service/database/Engine.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -29,13 +29,13 @@ class Engine:
		engine = sqlalchemy.create_engine(
		crdb_uri, connect_args={'application_name': APP_NAME}, echo=ECHO, future=True)
		except: # pylint: disable=bare-except # pragma: no cover
		LOGGER.exception('Failed to connect to database: {:s}'.format(crdb_uri))
		LOGGER.exception('Failed to connect to database: {:s}'.format(str(crdb_uri)))
		return None

		try:
		Engine.create_database(engine)
		except: # pylint: disable=bare-except # pragma: no cover
		LOGGER.exception('Failed to check/create to database: {:s}'.format(engine.url))
		LOGGER.exception('Failed to check/create to database: {:s}'.format(str(crdb_uri)))
		return None

		return engine