installers/charm/osm-nbi/src/legacy_interfaces.py

#!/usr/bin/env python3
# Copyright 2022 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
# For those usages not covered by the Apache License, Version 2.0 please
# contact: legal@canonical.com
#
# To get in touch with the maintainers, please contact:
# osm-charmers@lists.launchpad.net
#
# flake8: noqa

import ops


class BaseRelationClient(ops.framework.Object):
    """Requires side of a Kafka Endpoint"""

    def __init__(
        self,
        charm: ops.charm.CharmBase,
        relation_name: str,
        mandatory_fields: list = [],
    ):
        super().__init__(charm, relation_name)
        self.relation_name = relation_name
        self.mandatory_fields = mandatory_fields
        self._update_relation()

    def get_data_from_unit(self, key: str):
        if not self.relation:
            # This update relation doesn't seem to be needed, but I added it because apparently
            # the data is empty in the unit tests.
            # In reality, the constructor is called in every hook.
            # In the unit tests when doing an update_relation_data, apparently it is not called.
            self._update_relation()
        if self.relation:
            for unit in self.relation.units:
                data = self.relation.data[unit].get(key)
                if data:
                    return data

    def get_data_from_app(self, key: str):
        if not self.relation or self.relation.app not in self.relation.data:
            # This update relation doesn't seem to be needed, but I added it because apparently
            # the data is empty in the unit tests.
            # In reality, the constructor is called in every hook.
            # In the unit tests when doing an update_relation_data, apparently it is not called.
            self._update_relation()
        if self.relation and self.relation.app in self.relation.data:
            data = self.relation.data[self.relation.app].get(key)
            if data:
                return data

    def is_missing_data_in_unit(self):
        return not all([self.get_data_from_unit(field) for field in self.mandatory_fields])

    def is_missing_data_in_app(self):
        return not all([self.get_data_from_app(field) for field in self.mandatory_fields])

    def _update_relation(self):
        self.relation = self.framework.model.get_relation(self.relation_name)


class KeystoneClient(BaseRelationClient):
    """Requires side of a Keystone Endpoint"""

    mandatory_fields = [
        "host",
        "port",
        "user_domain_name",
        "project_domain_name",
        "username",
        "password",
        "service",
        "keystone_db_password",
        "region_id",
        "admin_username",
        "admin_password",
        "admin_project_name",
    ]

    def __init__(self, charm: ops.charm.CharmBase, relation_name: str):
        super().__init__(charm, relation_name, self.mandatory_fields)

    @property
    def host(self):
        return self.get_data_from_app("host")

    @property
    def port(self):
        return self.get_data_from_app("port")

    @property
    def user_domain_name(self):
        return self.get_data_from_app("user_domain_name")

    @property
    def project_domain_name(self):
        return self.get_data_from_app("project_domain_name")

    @property
    def username(self):
        return self.get_data_from_app("username")

    @property
    def password(self):
        return self.get_data_from_app("password")

    @property
    def service(self):
        return self.get_data_from_app("service")

    @property
    def keystone_db_password(self):
        return self.get_data_from_app("keystone_db_password")

    @property
    def region_id(self):
        return self.get_data_from_app("region_id")

    @property
    def admin_username(self):
        return self.get_data_from_app("admin_username")

    @property
    def admin_password(self):
        return self.get_data_from_app("admin_password")

    @property
    def admin_project_name(self):
        return self.get_data_from_app("admin_project_name")


class MongoClient(BaseRelationClient):
    """Requires side of a Mongo Endpoint"""

    mandatory_fields_mapping = {
        "reactive": ["connection_string"],
        "ops": ["replica_set_uri", "replica_set_name"],
    }

    def __init__(self, charm: ops.charm.CharmBase, relation_name: str):
        super().__init__(charm, relation_name, mandatory_fields=[])

    @property
    def connection_string(self):
        if self.is_opts():
            replica_set_uri = self.get_data_from_unit("replica_set_uri")
            replica_set_name = self.get_data_from_unit("replica_set_name")
            return f"{replica_set_uri}?replicaSet={replica_set_name}"
        else:
            return self.get_data_from_unit("connection_string")

    def is_opts(self):
        return not self.is_missing_data_in_unit_ops()

    def is_missing_data_in_unit(self):
        return self.is_missing_data_in_unit_ops() and self.is_missing_data_in_unit_reactive()

    def is_missing_data_in_unit_ops(self):
        return not all(
            [self.get_data_from_unit(field) for field in self.mandatory_fields_mapping["ops"]]
        )

    def is_missing_data_in_unit_reactive(self):
        return not all(
            [self.get_data_from_unit(field) for field in self.mandatory_fields_mapping["reactive"]]
        )


class PrometheusClient(BaseRelationClient):
    """Requires side of a Prometheus Endpoint"""

    mandatory_fields = ["hostname", "port"]

    def __init__(self, charm: ops.charm.CharmBase, relation_name: str):
        super().__init__(charm, relation_name, self.mandatory_fields)

    @property
    def hostname(self):
        return self.get_data_from_app("hostname")

    @property
    def port(self):
        return self.get_data_from_app("port")

    @property
    def user(self):
        return self.get_data_from_app("user")

    @property
    def password(self):
        return self.get_data_from_app("password")