How to use the ops.framework.EventSource function in ops

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.

github canonical / operator / test / test_framework.py View on Github external
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")
github canonical / operator / test / test_framework.py View on Github external
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()
github canonical / operator / test / test_framework.py View on Github external
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)
github canonical / operator / test / test_framework.py View on Github external
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 again
github canonical / operator / test / test_charm.py View on Github external
os.environ.clear()
            os.environ.update(env)
        self.addCleanup(restore_env, os.environ.copy())

        os.environ['PATH'] = "{}:{}".format(Path(__file__).parent / 'bin', os.environ['PATH'])
        os.environ['JUJU_UNIT_NAME'] = 'local/0'

        self.tmpdir = Path(tempfile.mkdtemp())
        self.addCleanup(shutil.rmtree, str(self.tmpdir))
        self.meta = CharmMeta()

        class CustomEvent(EventBase):
            pass

        class TestCharmEvents(CharmEvents):
            custom = EventSource(CustomEvent)

        # Relations events are defined dynamically and modify the class attributes.
        # We use a subclass temporarily to prevent these side effects from leaking.
        CharmBase.on = TestCharmEvents()

        def cleanup():
            CharmBase.on = CharmEvents()
        self.addCleanup(cleanup)
github canonical / operator / test / test_framework.py View on Github external
def test_bad_sig_observer(self):

        class MyEvent(EventBase):
            pass

        class MyNotifier(Object):
            foo = EventSource(MyEvent)
            bar = EventSource(MyEvent)
            baz = EventSource(MyEvent)
            qux = EventSource(MyEvent)

        class MyObserver(Object):
            def _on_foo(self):
                assert False, 'should not be reached'

            def _on_bar(self, event, extra):
                assert False, 'should not be reached'

            def _on_baz(self, event, extra=None, *, k):
                assert False, 'should not be reached'

            def _on_qux(self, event, extra=None):
                assert False, 'should not be reached'
github canonical / operator / ops / charm.py View on Github external
Called when new storage is available for the charm to use.
    """


class StorageDetachingEvent(StorageEvent):
    """Represents the `storage-detaching` hook from Juju.

    Called when storage a charm has been using is going away.
    """


class CharmEvents(ObjectEvents):
    """The events that are generated by Juju in response to the lifecycle of an application."""

    install = EventSource(InstallEvent)
    start = EventSource(StartEvent)
    stop = EventSource(StopEvent)
    remove = EventSource(RemoveEvent)
    update_status = EventSource(UpdateStatusEvent)
    config_changed = EventSource(ConfigChangedEvent)
    upgrade_charm = EventSource(UpgradeCharmEvent)
    pre_series_upgrade = EventSource(PreSeriesUpgradeEvent)
    post_series_upgrade = EventSource(PostSeriesUpgradeEvent)
    leader_elected = EventSource(LeaderElectedEvent)
    leader_settings_changed = EventSource(LeaderSettingsChangedEvent)
    collect_metrics = EventSource(CollectMetricsEvent)


class CharmBase(Object):
    """Base class that represents the Charm overall.

    Usually this initialization is done by ops.main.main() rather than Charm authors
github canonical / operator / ops / charm.py View on Github external
class StorageDetachingEvent(StorageEvent):
    """Represents the `storage-detaching` hook from Juju.

    Called when storage a charm has been using is going away.
    """


class CharmEvents(ObjectEvents):
    """The events that are generated by Juju in response to the lifecycle of an application."""

    install = EventSource(InstallEvent)
    start = EventSource(StartEvent)
    stop = EventSource(StopEvent)
    remove = EventSource(RemoveEvent)
    update_status = EventSource(UpdateStatusEvent)
    config_changed = EventSource(ConfigChangedEvent)
    upgrade_charm = EventSource(UpgradeCharmEvent)
    pre_series_upgrade = EventSource(PreSeriesUpgradeEvent)
    post_series_upgrade = EventSource(PostSeriesUpgradeEvent)
    leader_elected = EventSource(LeaderElectedEvent)
    leader_settings_changed = EventSource(LeaderSettingsChangedEvent)
    collect_metrics = EventSource(CollectMetricsEvent)


class CharmBase(Object):
    """Base class that represents the Charm overall.

    Usually this initialization is done by ops.main.main() rather than Charm authors
    directly instantiating a Charm.
github canonical / operator / ops / charm.py View on Github external
"""


class StorageDetachingEvent(StorageEvent):
    """Represents the `storage-detaching` hook from Juju.

    Called when storage a charm has been using is going away.
    """


class CharmEvents(ObjectEvents):
    """The events that are generated by Juju in response to the lifecycle of an application."""

    install = EventSource(InstallEvent)
    start = EventSource(StartEvent)
    stop = EventSource(StopEvent)
    remove = EventSource(RemoveEvent)
    update_status = EventSource(UpdateStatusEvent)
    config_changed = EventSource(ConfigChangedEvent)
    upgrade_charm = EventSource(UpgradeCharmEvent)
    pre_series_upgrade = EventSource(PreSeriesUpgradeEvent)
    post_series_upgrade = EventSource(PostSeriesUpgradeEvent)
    leader_elected = EventSource(LeaderElectedEvent)
    leader_settings_changed = EventSource(LeaderSettingsChangedEvent)
    collect_metrics = EventSource(CollectMetricsEvent)


class CharmBase(Object):
    """Base class that represents the Charm overall.

    Usually this initialization is done by ops.main.main() rather than Charm authors
    directly instantiating a Charm.
github canonical / operator / ops / charm.py View on Github external
Called when storage a charm has been using is going away.
    """


class CharmEvents(ObjectEvents):
    """The events that are generated by Juju in response to the lifecycle of an application."""

    install = EventSource(InstallEvent)
    start = EventSource(StartEvent)
    stop = EventSource(StopEvent)
    remove = EventSource(RemoveEvent)
    update_status = EventSource(UpdateStatusEvent)
    config_changed = EventSource(ConfigChangedEvent)
    upgrade_charm = EventSource(UpgradeCharmEvent)
    pre_series_upgrade = EventSource(PreSeriesUpgradeEvent)
    post_series_upgrade = EventSource(PostSeriesUpgradeEvent)
    leader_elected = EventSource(LeaderElectedEvent)
    leader_settings_changed = EventSource(LeaderSettingsChangedEvent)
    collect_metrics = EventSource(CollectMetricsEvent)


class CharmBase(Object):
    """Base class that represents the Charm overall.

    Usually this initialization is done by ops.main.main() rather than Charm authors
    directly instantiating a Charm.

    Args:
        framework: The framework responsible for managing the Model and events for this
            Charm.
        key: Ignored; will remove after deprecation period of the signature change.