2 # Copyright 2021 Canonical Ltd.
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
16 # For those usages not covered by the Apache License, Version 2.0 please
17 # contact: legal@canonical.com
19 # To get in touch with the maintainers, please contact:
20 # osm-charmers@lists.launchpad.net
24 import zaza
.model
as model
25 from pymongo
import MongoClient
29 mongo_uri
= "mongodb://"
30 mongo_units
= model
.get_status().applications
["mongodb-k8s"]["units"]
31 for i
, unit_name
in enumerate(mongo_units
.keys()):
34 unit_ip
= mongo_units
[unit_name
]["address"]
35 unit_port
= mongo_units
[unit_name
]["opened-ports"][0].split("/")[0]
36 mongo_uri
+= "{}:{}".format(unit_ip
, unit_port
)
41 class BasicDeployment(unittest
.TestCase
):
42 def test_get_mongo_uri(self
):
45 def test_mongodb_connection(self
):
46 mongo_uri
= get_mongo_uri()
47 client
= MongoClient(mongo_uri
)
50 def test_mongodb_create_empty_database_collection(self
):
51 mongo_uri
= get_mongo_uri()
52 client
= MongoClient(mongo_uri
)
53 DB_NAME
= "test_database"
54 COLLECTION_NAME
= "test_collection"
57 _
= client
.list_database_names()
59 collection
= db
[COLLECTION_NAME
]
60 _
= db
.list_collection_names()
64 id = collection
.insert_one(data
)
66 for x
in collection
.find({"_id": id.inserted_id
}):
67 self
.assertEqual(id.inserted_id
, x
["_id"])
69 def test_mongodb_insert_one(self
):
70 mongo_uri
= get_mongo_uri()
71 client
= MongoClient(mongo_uri
)
72 DB_NAME
= "test_database"
73 COLLECTION_NAME
= "test_collection"
76 _
= client
.list_database_names()
78 collection
= db
[COLLECTION_NAME
]
79 _
= db
.list_collection_names()
82 "name": "Canonical LTD",
83 "address": "5th Floor of the Blue Fin Building",
86 id = collection
.insert_one(data
)
88 for x
in collection
.find({"_id": id.inserted_id
}):
89 self
.assertEqual(id.inserted_id
, x
["_id"])
91 def test_mongodb_insert_many(self
):
92 mongo_uri
= get_mongo_uri()
93 client
= MongoClient(mongo_uri
)
94 DB_NAME
= "test_database"
95 COLLECTION_NAME
= "test_collection"
98 _
= client
.list_database_names()
100 collection
= db
[COLLECTION_NAME
]
101 _
= db
.list_collection_names()
104 {"name": "Amy", "address": "Apple st 652"},
105 {"name": "Hannah", "address": "Mountain 21"},
106 {"name": "Michael", "address": "Valley 345"},
107 {"name": "Sandy", "address": "Ocean blvd 2"},
108 {"name": "Betty", "address": "Green Grass 1"},
109 {"name": "Richard", "address": "Sky st 331"},
110 {"name": "Susan", "address": "One way 98"},
111 {"name": "Vicky", "address": "Yellow Garden 2"},
112 {"name": "Ben", "address": "Park Lane 38"},
113 {"name": "William", "address": "Central st 954"},
114 {"name": "Chuck", "address": "Main Road 989"},
115 {"name": "Viola", "address": "Sideway 1633"},
118 ids
= collection
.insert_many(data
)
120 for id in ids
.inserted_ids
:
121 x
= collection
.find_one({"_id": id})
122 self
.assertEqual(x
["_id"], id)