installers/charm/pla/tests/test_charm.py

   1 #!/usr/bin/env python3
   2 # Copyright 2021 Canonical Ltd.
   3 #
   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
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 ##
  22
  23 import sys
  24 from typing import NoReturn
  25 import unittest
  26
  27
  28 from charm import PlaCharm
  29 from ops.model import ActiveStatus, BlockedStatus
  30 from ops.testing import Harness
  31
  32
  33 class TestCharm(unittest.TestCase):
  34     """Pla Charm unit tests."""
  35
  36     def setUp(self) -> NoReturn:
  37         """Test setup"""
  38         self.image_info = sys.modules["oci_image"].OCIImageResource().fetch()
  39         self.harness = Harness(PlaCharm)
  40         self.harness.set_leader(is_leader=True)
  41         self.harness.begin()
  42         self.config = {
  43             "log_level": "INFO",
  44             "mongodb_uri": "",
  45         }
  46         self.harness.update_config(self.config)
  47
  48     def test_config_changed_no_relations(
  49         self,
  50     ) -> NoReturn:
  51         """Test ingress resources without HTTP."""
  52
  53         self.harness.charm.on.config_changed.emit()
  54
  55         # Assertions
  56         self.assertIsInstance(self.harness.charm.unit.status, BlockedStatus)
  57         self.assertTrue(
  58             all(
  59                 relation in self.harness.charm.unit.status.message
  60                 for relation in ["mongodb", "kafka"]
  61             )
  62         )
  63
  64     def test_config_changed_non_leader(
  65         self,
  66     ) -> NoReturn:
  67         """Test ingress resources without HTTP."""
  68         self.harness.set_leader(is_leader=False)
  69         self.harness.charm.on.config_changed.emit()
  70
  71         # Assertions
  72         self.assertIsInstance(self.harness.charm.unit.status, ActiveStatus)
  73
  74     def test_with_relations_and_mongodb_config(
  75         self,
  76     ) -> NoReturn:
  77         "Test with relations and mongodb config (internal)"
  78         self.initialize_kafka_relation()
  79         self.initialize_mongo_config()
  80         # Verifying status
  81         self.assertNotIsInstance(self.harness.charm.unit.status, BlockedStatus)
  82
  83     def test_with_relations(
  84         self,
  85     ) -> NoReturn:
  86         "Test with relations (internal)"
  87         self.initialize_kafka_relation()
  88         self.initialize_mongo_relation()
  89         # Verifying status
  90         self.assertNotIsInstance(self.harness.charm.unit.status, BlockedStatus)
  91
  92     def test_exception_mongodb_relation_and_config(
  93         self,
  94     ) -> NoReturn:
  95         "Test with relation and config for Mongodb. Test must fail"
  96         self.initialize_mongo_relation()
  97         self.initialize_mongo_config()
  98         # Verifying status
  99         self.assertIsInstance(self.harness.charm.unit.status, BlockedStatus)
 100
 101     def initialize_kafka_relation(self):
 102         kafka_relation_id = self.harness.add_relation("kafka", "kafka")
 103         self.harness.add_relation_unit(kafka_relation_id, "kafka/0")
 104         self.harness.update_relation_data(
 105             kafka_relation_id, "kafka", {"host": "kafka", "port": 9092}
 106         )
 107
 108     def initialize_mongo_config(self):
 109         self.harness.update_config({"mongodb_uri": "mongodb://mongo:27017"})
 110
 111     def initialize_mongo_relation(self):
 112         mongodb_relation_id = self.harness.add_relation("mongodb", "mongodb")
 113         self.harness.add_relation_unit(mongodb_relation_id, "mongodb/0")
 114         self.harness.update_relation_data(
 115             mongodb_relation_id,
 116             "mongodb/0",
 117             {"connection_string": "mongodb://mongo:27017"},
 118         )
 119
 120
 121 if __name__ == "__main__":
 122     unittest.main()