installers/charm/kafka/tests/test_charm.py

   1 #!/usr/bin/env python3
   2 # Copyright 2020 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 from typing import NoReturn
  24 import unittest
  25 from unittest.mock import patch, PropertyMock
  26
  27 from charm import KafkaCharm
  28 from ops.model import ActiveStatus, BlockedStatus
  29 from ops.testing import Harness
  30
  31
  32 class TestCharm(unittest.TestCase):
  33     """Kafka Charm unit tests."""
  34
  35     def setUp(
  36         self,
  37     ) -> NoReturn:
  38         """Test setup"""
  39         self.harness = Harness(KafkaCharm)
  40         self.harness.set_leader(is_leader=True)
  41         self.harness.begin()
  42         self.config = {"num_partitions": 1}
  43         self.harness.update_config(self.config)
  44
  45     def test_config_changed_no_relations(self) -> NoReturn:
  46         """Test config changed without relations."""
  47         self.harness.charm.on.config_changed.emit()
  48         self.assertIsInstance(self.harness.charm.unit.status, BlockedStatus)
  49
  50     def test_config_changed_non_leader(self) -> NoReturn:
  51         """Test config changed without relations (non-leader)."""
  52         self.harness.set_leader(is_leader=False)
  53         self.harness.charm.on.config_changed.emit()
  54
  55         # Assertions
  56         self.assertIsInstance(self.harness.charm.unit.status, ActiveStatus)
  57
  58     @patch("charm.KafkaCharm.num_units", new_callable=PropertyMock)
  59     def test_with_relations_kafka(self, mock_num_units) -> NoReturn:
  60         "Test with relations (kafka)"
  61         mock_num_units.return_value = 1
  62
  63         # Initializing the kafka relation
  64         zookeeper_relation_id = self.harness.add_relation("zookeeper", "zookeeper")
  65         self.harness.add_relation_unit(zookeeper_relation_id, "zookeeper/0")
  66         self.harness.update_relation_data(
  67             zookeeper_relation_id, "zookeeper", {"zookeeper_uri": "zk-uri"}
  68         )
  69
  70         # Verifying status
  71         self.assertNotIsInstance(self.harness.charm.unit.status, BlockedStatus)
  72
  73
  74 if __name__ == "__main__":
  75     unittest.main()