Directory structure cleanup under 'hub'

src/omnipy/engine/local.py

@@ -8,7 +8,7 @@
                                       FuncFlowRunnerEngine,
                                       LinearFlowRunnerEngine,
                                       TaskRunnerEngine)
-from omnipy.hub.entry import RuntimeEntryPublisher
+from omnipy.util.publisher import RuntimeEntryPublisher
 
 
 @dataclass

src/omnipy/hub/log/root_log.py

@@ -6,9 +6,9 @@
 
 from omnipy.api.protocols.public.config import IsRootLogConfig
 from omnipy.config.root_log import RootLogConfig
-from omnipy.hub.entry import RuntimeEntryPublisher
 from omnipy.hub.log.handlers import DailyRotatingFileHandler
 from omnipy.util.helpers import get_datetime_format
+from omnipy.util.publisher import RuntimeEntryPublisher
 
 
 @dataclass

src/omnipy/hub/runtime.py

@@ -21,11 +21,11 @@
 from omnipy.data.data_class_creator import DataClassBase
 from omnipy.data.serializer import SerializerRegistry
 from omnipy.engine.local import LocalRunner, LocalRunnerConfigEntryPublisher
-from omnipy.hub.entry import DataPublisher, RuntimeEntryPublisher
 from omnipy.hub.log.root_log import RootLogConfigEntryPublisher, RootLogObjects
 from omnipy.hub.registry import RunStateRegistry
 from omnipy.modules.prefect.engine.prefect import PrefectEngine, PrefectEngineConfigEntryPublisher
 from omnipy.util.helpers import called_from_omnipy_tests
+from omnipy.util.publisher import DataPublisher, RuntimeEntryPublisher
 
 
 def _job_creator_factory():

src/omnipy/modules/prefect/engine/prefect.py

@@ -10,8 +10,8 @@
                                       FuncFlowRunnerEngine,
                                       LinearFlowRunnerEngine,
                                       TaskRunnerEngine)
-from omnipy.hub.entry import RuntimeEntryPublisher
 from omnipy.util.helpers import resolve
+from omnipy.util.publisher import RuntimeEntryPublisher
 
 from .. import prefect_flow, prefect_task, PrefectFlow, PrefectTask, task_input_hash
 

src/omnipy/util/publisher.py

@@ -2,6 +2,8 @@
 from dataclasses import dataclass, field
 from typing import Any, Callable, DefaultDict
 
+from omnipy.api.protocols.public.hub import IsRuntime
+
 
 def _subscribers_factory():
     return defaultdict(list)
@@ -30,3 +32,14 @@ def __setattr__(self, key, value):
         if key in self._subscriptions:
             for callback_fun in self._subscriptions[key]:
                 callback_fun(value)
+
+
+@dataclass
+class RuntimeEntryPublisher(DataPublisher):
+    _back: IsRuntime | None = field(default=None, init=False, repr=False)
+
+    def __setattr__(self, key, value):
+        super().__setattr__(key, value)
+
+        if hasattr(self, key) and not key.startswith('_') and self._back is not None:
+            self._back.reset_subscriptions()

tests/hub/log/test_log_mixin.py

@@ -10,7 +10,7 @@
 from omnipy.hub.log.mixin import LogMixin
 from omnipy.util.mixin import DynamicMixinAcceptor
 
-from .helpers.functions import assert_log_line_from_stream, format_datetime_obj
+from ..helpers.functions import assert_log_line_from_stream, format_datetime_obj
 
 
 @pc.case(id='my_class_as_regular_log_mixin_subclass')

tests/hub/log/test_root_log.py

@@ -11,14 +11,13 @@
 import pytest
 
 from omnipy.api.protocols.public.hub import IsRuntime
-from omnipy.api.typedefs import LocaleType
 from omnipy.config.root_log import RootLogConfig
 from omnipy.hub.log.root_log import RootLogObjects
 from omnipy.util.helpers import get_datetime_format
 
-from .helpers.functions import (assert_log_line_from_stream,
-                                format_datetime_obj,
-                                read_log_line_from_stream)
+from ..helpers.functions import (assert_log_line_from_stream,
+                                 format_datetime_obj,
+                                 read_log_line_from_stream)
 
 
 def _assert_root_log_config_default(root_log: RootLogConfig, dir_path: Path):

tests/hub/registry/conftest.py

@@ -0,0 +1 @@
+

tests/hub/test_registry.py → tests/hub/registry/test_registry.py

@@ -11,7 +11,62 @@
 from omnipy.api.protocols.public.hub import IsRuntime
 from omnipy.hub.registry import RunStateRegistry
 
-from .log.helpers.functions import read_log_lines_from_stream
+from ...engine.helpers.mocks import (MockDagFlow,
+                                     MockDagFlowTemplate,
+                                     MockJobRunnerSubclass,
+                                     MockTask,
+                                     MockTaskTemplate)
+from ..helpers.functions import read_log_lines_from_stream
+
+
+@pytest.fixture(scope='module')
+def task_template_a() -> MockTaskTemplate:
+    MockTaskTemplate.job_creator.engine = MockJobRunnerSubclass()
+
+    @MockTaskTemplate(name='a')
+    def concat_a(s: str) -> str:
+        return s + 'a'
+
+    return concat_a
+
+
+@pytest.fixture(scope='module')
+def task_template_b() -> MockTaskTemplate:
+    MockTaskTemplate.job_creator.engine = MockJobRunnerSubclass()
+
+    @MockTaskTemplate(name='b')
+    def concat_b(s: str) -> str:
+        return s + 'b'
+
+    return concat_b
+
+
+@pytest.fixture(scope='module')
+def task_a(task_template_a) -> MockTask:
+    return task_template_a.apply()
+
+
+@pytest.fixture(scope='module')
+def task_b(task_template_b) -> MockTask:
+    return task_template_b.apply()
+
+
+@pytest.fixture(scope='module')
+def dag_flow_a(task_template_a, task_template_b) -> MockDagFlow:
+    @MockDagFlowTemplate(task_template_a, task_template_b, name='a')
+    def concat_a(s: str) -> str:
+        ...
+
+    return concat_a.apply()
+
+
+@pytest.fixture(scope='module')
+def dag_flow_b(task_template_a, task_template_b) -> MockDagFlow:
+    @MockDagFlowTemplate(task_template_a, task_template_b, name='b')
+    def concat_b(s: str) -> str:
+        ...
+
+    return concat_b.apply()
 
 
 def test_job_state_transitions(