from typing import List, Optional from uaclient.api.api import APIEndpoint from uaclient.api.data_types import AdditionalInfo from uaclient.api.u.pro.status.is_attached.v1 import _is_attached from uaclient.config import UAConfig from uaclient.data_types import ( BoolDataValue, DataObject, Field, StringDataValue, data_list, ) class EnabledService(DataObject): fields = [ Field("name", StringDataValue), Field("variant_enabled", BoolDataValue), Field("variant_name", StringDataValue, False), ] def __init__( self, *, name: str, variant_enabled: bool = False, variant_name: Optional[str] = None ): self.name = name self.variant_enabled = variant_enabled self.variant_name = variant_name class EnabledServicesResult(DataObject, AdditionalInfo): fields = [ Field("enabled_services", data_list(EnabledService)), ] def __init__(self, *, enabled_services: List[EnabledService]): self.enabled_services = enabled_services def enabled_services() -> EnabledServicesResult: return _enabled_services(UAConfig()) def _enabled_services(cfg: UAConfig) -> EnabledServicesResult: from uaclient.entitlements import ENTITLEMENT_CLASSES from uaclient.entitlements.entitlement_status import UserFacingStatus if not _is_attached(cfg).is_attached: return EnabledServicesResult(enabled_services=[]) enabled_services = [] # type: List[EnabledService] for ent_cls in ENTITLEMENT_CLASSES: ent = ent_cls(cfg) if ent.user_facing_status()[0] == UserFacingStatus.ACTIVE: enabled_service = EnabledService(name=ent.name) for _, variant_cls in ent.variants.items(): variant = variant_cls(cfg) if variant.user_facing_status()[0] == UserFacingStatus.ACTIVE: enabled_service = EnabledService( name=ent.name, variant_enabled=True, variant_name=variant.variant_name, ) break enabled_services.append(enabled_service) return EnabledServicesResult( enabled_services=sorted(enabled_services, key=lambda x: x.name) ) endpoint = APIEndpoint( version="v1", name="EnabledServices", fn=_enabled_services, options_cls=None, )