from defence360agent.rpc_tools import ValidationError from defence360agent.rpc_tools.lookup import RootEndpoints, bind from defence360agent.utils import Scope from im360.subsys.features import get_applicable_features from defence360agent.subsys.features.abstract_feature import ( FeatureError, FeatureStatus, ) class FeaturesEndpoints(RootEndpoints): SCOPE = Scope.IM360 @bind("features", "list") async def feature_list(self): return {"items": list(get_applicable_features().keys())} @bind("features", "install") async def feature_install(self, name): return await self.call("install", name) @bind("features", "remove") async def feature_remove(self, name): return await self.call("remove", name) @bind("features", "status") async def feature_status(self, name): return await self.call("status", name) async def call(self, action, name): try: _cls = get_applicable_features()[name] except KeyError: raise ValidationError("Feature is not available: {}".format(name)) else: if _cls is None: raise ValidationError(f"No implementation for feature: {name}") try: feature = await _cls(sink=self._sink).init() return await getattr(feature, action)() except FeatureError as e: # implemented new format for status, but not for install/remove yet # also will be nice to move log file path to status response # because now, if status returned "installing", UI have to call # "install" (!) to get log file path if action == "status": return { "items": { "status": FeatureStatus.ERROR, "message": str(e), } } else: raise ValidationError(str(e))