installers/charm/mongodb-k8s/tests/basic_deployment.py

#!/usr/bin/python3
# Copyright 2021 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
##

import unittest
import zaza.model as model
from pymongo import MongoClient


def get_mongo_uri():
    mongo_uri = "mongodb://"
    mongo_units = model.get_status().applications["mongodb-k8s"]["units"]
    for i, unit_name in enumerate(mongo_units.keys()):
        if i:
            mongo_uri += ","
        unit_ip = mongo_units[unit_name]["address"]
        unit_port = mongo_units[unit_name]["opened-ports"][0].split("/")[0]
        mongo_uri += "{}:{}".format(unit_ip, unit_port)

    return mongo_uri


class BasicDeployment(unittest.TestCase):
    def test_get_mongo_uri(self):
        get_mongo_uri()

    def test_mongodb_connection(self):
        mongo_uri = get_mongo_uri()
        client = MongoClient(mongo_uri)
        client.server_info()

    def test_mongodb_create_empty_database_collection(self):
        mongo_uri = get_mongo_uri()
        client = MongoClient(mongo_uri)
        DB_NAME = "test_database"
        COLLECTION_NAME = "test_collection"

        db = client[DB_NAME]
        _ = client.list_database_names()

        collection = db[COLLECTION_NAME]
        _ = db.list_collection_names()

        data = {}

        id = collection.insert_one(data)

        for x in collection.find({"_id": id.inserted_id}):
            self.assertEqual(id.inserted_id, x["_id"])

    def test_mongodb_insert_one(self):
        mongo_uri = get_mongo_uri()
        client = MongoClient(mongo_uri)
        DB_NAME = "test_database"
        COLLECTION_NAME = "test_collection"

        db = client[DB_NAME]
        _ = client.list_database_names()

        collection = db[COLLECTION_NAME]
        _ = db.list_collection_names()

        data = {
            "name": "Canonical LTD",
            "address": "5th Floor of the Blue Fin Building",
        }

        id = collection.insert_one(data)

        for x in collection.find({"_id": id.inserted_id}):
            self.assertEqual(id.inserted_id, x["_id"])

    def test_mongodb_insert_many(self):
        mongo_uri = get_mongo_uri()
        client = MongoClient(mongo_uri)
        DB_NAME = "test_database"
        COLLECTION_NAME = "test_collection"

        db = client[DB_NAME]
        _ = client.list_database_names()

        collection = db[COLLECTION_NAME]
        _ = db.list_collection_names()

        data = [
            {"name": "Amy", "address": "Apple st 652"},
            {"name": "Hannah", "address": "Mountain 21"},
            {"name": "Michael", "address": "Valley 345"},
            {"name": "Sandy", "address": "Ocean blvd 2"},
            {"name": "Betty", "address": "Green Grass 1"},
            {"name": "Richard", "address": "Sky st 331"},
            {"name": "Susan", "address": "One way 98"},
            {"name": "Vicky", "address": "Yellow Garden 2"},
            {"name": "Ben", "address": "Park Lane 38"},
            {"name": "William", "address": "Central st 954"},
            {"name": "Chuck", "address": "Main Road 989"},
            {"name": "Viola", "address": "Sideway 1633"},
        ]

        ids = collection.insert_many(data)

        for id in ids.inserted_ids:
            x = collection.find_one({"_id": id})
            self.assertEqual(x["_id"], id)