Load patches in static definition

src/saturn_engine/worker_manager/api/jobs.py

@@ -67,6 +67,8 @@ def update_job(job_name: str) -> Json[UpdateResponse]:
 def post_sync() -> Json[JobsSyncResponse]:
     """Create jobs that are due to be scheduled."""
     with session_scope() as session:
+        # We reset static definition at each jobs sync
+        current_app.saturn.load_static_definition(session=session)
         sync_jobs(
             static_definitions=current_app.saturn.static_definitions,
             session=session,

src/saturn_engine/worker_manager/context.py

@@ -24,21 +24,23 @@ def static_definitions(self) -> StaticDefinitions:
         return self._static_definitions
 
     def load_static_definition(self, session: AnySession) -> None:
-        self._static_definitions = _load_static_definition(
-            session=session, config=self.config
-        )
+        static_definition = _load_static_definition(session=session, config=self.config)
+        if static_definition:
+            self._static_definitions = static_definition
 
 
 def _load_static_definition(
     config: WorkerManagerConfig, session: AnySession
-) -> StaticDefinitions:
+) -> StaticDefinitions | None:
     """
     Static definitions contain objects defined in a declarative configuration:
     - Inventories
     - Topics
     - Jobs
     - JobDefinitions
     """
+    if not config.static_definitions_directories:
+        return None
     patches = topologies_store.get_patches(session=session)
     definitions = load_definitions_from_paths(
         config.static_definitions_directories, patches=patches

src/saturn_engine/worker_manager/server.py

@@ -1,5 +1,7 @@
 from typing import Optional
 
+from pytest import Config
+
 from saturn_engine.config import default_config_with_env
 from saturn_engine.database import create_all
 from saturn_engine.database import scoped_session
@@ -12,19 +14,22 @@
 
 
 def get_app(
-    config: Optional[dict] = None,
+    saturn_config: Config | None = None,
+    app_config: Optional[dict] = None,
 ) -> SaturnApp:
-    worker_manager_context = WorkerManagerContext(
-        config=default_config_with_env().c.worker_manager,
-    )
+    if saturn_config:
+        worker_manager_config = saturn_config.c.worker_manager
+    else:
+        worker_manager_config = default_config_with_env().c.worker_manager
+    worker_manager_context = WorkerManagerContext(config=worker_manager_config)
 
     app = SaturnApp(
         worker_manager_context,
         __name__,
     )
 
-    if config:
-        app.config.from_mapping(config)
+    if app_config:
+        app.config.from_mapping(app_config)
 
     from .api.inventories import bp as bp_inventories
     from .api.job_definitions import bp as bp_job_definitions

tests/config.py

@@ -1,6 +1,7 @@
 from saturn_engine.config import Env
 from saturn_engine.config import SaturnConfig
 from saturn_engine.config import ServicesManagerConfig
+from saturn_engine.config_definitions import WorkerManagerConfig
 
 
 class config(SaturnConfig):
@@ -16,3 +17,6 @@ class services_manager(ServicesManagerConfig):
 
     class job_state:
         auto_flush = False
+
+    class worker_manager(WorkerManagerConfig):
+        static_definitions_directories = []

tests/worker_manager/conftest.py

@@ -5,6 +5,7 @@
 from sqlalchemy.orm import Session
 
 from saturn_engine import database
+from saturn_engine.config import Config
 from saturn_engine.core import JobId
 from saturn_engine.core import api
 from saturn_engine.models import Base
@@ -96,11 +97,12 @@ def fake_job_definition(
 
 
 @pytest.fixture
-def app() -> t.Iterator[SaturnApp]:
+def app(config: Config) -> t.Iterator[SaturnApp]:
     app = worker_manager_server.get_app(
-        config={
+        app_config={
             "TESTING": True,
         },
+        saturn_config=config,
     )
     with app.app_context():
         Base.metadata.drop_all(bind=database.engine())