installers/charm/osm-ro/tests/unit/test_charm.py

   1 #!/usr/bin/env python3
   2 # Copyright 2022 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 # Learn more about testing at: https://juju.is/docs/sdk/testing
  23
  24 import pytest
  25 from ops.model import ActiveStatus, BlockedStatus
  26 from ops.testing import Harness
  27 from pytest_mock import MockerFixture
  28
  29 from charm import CharmError, OsmRoCharm, check_service_active
  30
  31 container_name = "ro"
  32 service_name = "ro"
  33
  34
  35 @pytest.fixture
  36 def harness(mocker: MockerFixture):
  37     mocker.patch("charm.KubernetesServicePatch", lambda x, y: None)
  38     harness = Harness(OsmRoCharm)
  39     harness.begin()
  40     harness.container_pebble_ready(container_name)
  41     yield harness
  42     harness.cleanup()
  43
  44
  45 def test_missing_relations(harness: Harness):
  46     harness.charm.on.config_changed.emit()
  47     assert type(harness.charm.unit.status) == BlockedStatus
  48     assert all(relation in harness.charm.unit.status.message for relation in ["mongodb", "kafka"])
  49
  50
  51 def test_ready(harness: Harness):
  52     _add_relations(harness)
  53     assert harness.charm.unit.status == ActiveStatus()
  54
  55
  56 def test_container_stops_after_relation_broken(harness: Harness):
  57     harness.charm.on[container_name].pebble_ready.emit(container_name)
  58     container = harness.charm.unit.get_container(container_name)
  59     relation_ids = _add_relations(harness)
  60     check_service_active(container, service_name)
  61     harness.remove_relation(relation_ids[0])
  62     with pytest.raises(CharmError):
  63         check_service_active(container, service_name)
  64
  65
  66 def test_ro_relation_joined(harness: Harness):
  67     harness.set_leader(True)
  68     _add_relations(harness)
  69     relation_id = harness.add_relation("ro", "lcm")
  70     harness.add_relation_unit(relation_id, "lcm/0")
  71     relation_data = harness.get_relation_data(relation_id, harness.charm.app.name)
  72     assert harness.charm.unit.status == ActiveStatus()
  73     assert relation_data == {"host": harness.charm.app.name, "port": "9090"}
  74
  75
  76 def test_certificates(harness: Harness):
  77     # aGVsbG8K: "hello\n"
  78     # aGVsbG8gYWdhaW4K: "hello again\n"
  79     _add_relations(harness)
  80     harness.update_config({"certificates": "cert1:aGVsbG8K,cert2:aGVsbG8gYWdhaW4K"})
  81     for cert_name, content in {"cert1": "hello\n", "cert2": "hello again\n"}.items():
  82         assert harness.charm.container.exists(f"/certs/{cert_name}")
  83         assert harness.charm.container.pull(f"/certs/{cert_name}").read() == content
  84
  85
  86 def _add_relations(harness: Harness):
  87     relation_ids = []
  88     # Add mongo relation
  89     relation_id = harness.add_relation("mongodb", "mongodb")
  90     harness.add_relation_unit(relation_id, "mongodb/0")
  91     harness.update_relation_data(
  92         relation_id,
  93         "mongodb",
  94         {"uris": "mongodb://:1234", "username": "user", "password": "password"},
  95     )
  96     relation_ids.append(relation_id)
  97     # Add kafka relation
  98     relation_id = harness.add_relation("kafka", "kafka")
  99     harness.add_relation_unit(relation_id, "kafka/0")
 100     harness.update_relation_data(relation_id, "kafka", {"host": "kafka", "port": "9092"})
 101     relation_ids.append(relation_id)
 102     return relation_ids