How to use ops - 10 common examples
To help you get started, we’ve selected a few ops examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
canonical / operator / test / test_testing.py View on Github 
def test_update_relation_no_local_app_change_event(self):
# language=YAML
harness = Harness(CharmBase, meta='''
name: my-charm
requires:
db:
interface: pgsql
''')
self.addCleanup(harness.cleanup)
harness.begin()
harness.set_leader(False)
helper = DBRelationChangedHelper(harness.charm, "helper")
rel_id = harness.add_relation('db', 'postgresql')
# TODO: remove this as soon as https://github.com/canonical/operator/issues/175 is fixed.
harness.add_relation_unit(rel_id, 'postgresql/0')
self.assertEqual(helper.changes, [])
harness.update_relation_data(rel_id, 'my-charm', {'new': 'value'})
rel = harness.charm.model.get_relation('db')def test_relation_set_deletes(self):
harness = Harness(CharmBase, meta='''
name: test-charm
requires:
db:
interface: pgsql
''')
self.addCleanup(harness.cleanup)
harness.begin()
harness.set_leader(False)
rel_id = harness.add_relation('db', 'postgresql')
harness.update_relation_data(rel_id, 'test-charm/0', {'foo': 'bar'})
harness.add_relation_unit(rel_id, 'postgresql/0')
rel = harness.charm.model.get_relation('db', rel_id)
del rel.data[harness.charm.model.unit]['foo']
self.assertEqual({}, harness.get_relation_data(rel_id, 'test-charm/0'))def test_add_relation_and_unit(self):
harness = Harness(CharmBase, meta='''
name: test-app
requires:
db:
interface: pgsql
''')
self.addCleanup(harness.cleanup)
rel_id = harness.add_relation('db', 'postgresql')
self.assertIsInstance(rel_id, int)
harness.add_relation_unit(rel_id, 'postgresql/0')
harness.update_relation_data(rel_id, 'postgresql/0', {'foo': 'bar'})
backend = harness._backend
self.assertEqual(backend.relation_ids('db'), [rel_id])
self.assertEqual(backend.relation_list(rel_id), ['postgresql/0'])
self.assertEqual(
backend.relation_get(rel_id, 'postgresql/0', is_app=False),
{'foo': 'bar'})def test_defer_and_reemit(self):
framework = self.create_framework()
class MyEvent(EventBase):
pass
class MyNotifier1(Object):
a = EventSource(MyEvent)
b = EventSource(MyEvent)
class MyNotifier2(Object):
c = EventSource(MyEvent)
class MyObserver(Object):
def __init__(self, parent, key):
super().__init__(parent, key)
self.seen = []
self.done = {}
def on_any(self, event):
self.seen.append(event.handle.kind)
if not self.done.get(event.handle.kind):
event.defer()
pub1 = MyNotifier1(framework, "1")
pub2 = MyNotifier2(framework, "1")
obs1 = MyObserver(framework, "1")
obs2 = MyObserver(framework, "2")def test_defer_and_reemit(self):
framework = self.create_framework()
class MyEvent(EventBase):
pass
class MyNotifier1(Object):
a = EventSource(MyEvent)
b = EventSource(MyEvent)
class MyNotifier2(Object):
c = EventSource(MyEvent)
class MyObserver(Object):
def __init__(self, parent, key):
super().__init__(parent, key)
self.seen = []
self.done = {}
def on_any(self, event):
self.seen.append(event.handle.kind)
if not self.done.get(event.handle.kind):
event.defer()framework = self.create_framework()
class MyEvent(EventBase):
def __init__(self, handle, n):
super().__init__(handle)
self.my_n = n
def snapshot(self):
return {"My N!": self.my_n}
def restore(self, snapshot):
super().restore(snapshot)
self.my_n = snapshot["My N!"] + 1
class MyNotifier(Object):
foo = EventSource(MyEvent)
class MyObserver(Object):
def __init__(self, parent, key):
super().__init__(parent, key)
self.seen = []
def _on_foo(self, event):
self.seen.append("on_foo:{}={}".format(event.handle.kind, event.my_n))
event.defer()
pub = MyNotifier(framework, "1")
obs = MyObserver(framework, "1")
framework.observe(pub.foo, obs._on_foo)
pub.foo.emit(1)def test_weak_observer(self):
framework = self.create_framework()
observed_events = []
class MyEvent(EventBase):
pass
class MyEvents(ObjectEvents):
foo = EventSource(MyEvent)
class MyNotifier(Object):
on = MyEvents()
class MyObserver(Object):
def _on_foo(self, event):
observed_events.append("foo")
pub = MyNotifier(framework, "1")
obs = MyObserver(framework, "2")
framework.observe(pub.on.foo, obs._on_foo)
pub.on.foo.emit()
self.assertEqual(observed_events, ["foo"])
# Now delete the observer, and note that when we emit the event, it
# doesn't update the local slice againdef test_status_set_get_app(self):
harness = Harness(CharmBase, meta='''
name: app
''')
self.addCleanup(harness.cleanup)
backend = harness._backend
backend.status_set('blocked', 'message', is_app=True)
self.assertEqual(
backend.status_get(is_app=True),
{'status': 'blocked', 'message': 'message'})
self.assertEqual(
backend.status_get(is_app=False),
{'status': 'maintenance', 'message': ''})def test_add_resource_but_oci(self):
harness = Harness(CharmBase, meta='''
name: test-app
resources:
image:
type: oci-image
description: "Image to deploy."
''')
self.addCleanup(harness.cleanup)
with self.assertRaises(RuntimeError):
harness.add_resource('image', 'content')def test_relation_list_unknown_relation_id(self):
harness = Harness(CharmBase, meta='''
name: test-charm
''')
self.addCleanup(harness.cleanup)
backend = harness._backend
with self.assertRaises(RelationNotFoundError):
backend.relation_list(1234)
