Merge pull request #6 from insight-platform/fix-labels-parsing

Updating labels parsing and validation

Author: placccebo

src/pipeline_watchdog/config/config.py

@@ -4,6 +4,11 @@
 from typing import List, Optional
 
 
+def validate_container_labels(labels: List[List[str]]):
+    if not labels:
+        raise ValueError(f'Container labels cannot be empty.')
+
+
 class Action(Enum):
     STOP = 'stop'
     RESTART = 'restart'
@@ -28,6 +33,9 @@ class QueueConfig:
     container_labels: List[List[str]]
     """List of labels to filter the containers to which the action is applied."""
 
+    def __post_init__(self):
+        validate_container_labels(self.container_labels)
+
 
 @dataclass
 class FlowConfig:
@@ -48,6 +56,9 @@ class FlowConfig:
     container_labels: List[List[str]]
     """List of labels to filter the containers to which the action is applied."""
 
+    def __post_init__(self):
+        validate_container_labels(self.container_labels)
+
 
 @dataclass
 class WatchConfig:

src/pipeline_watchdog/config/parser.py

@@ -12,13 +12,15 @@ def __init__(self, config_path: str):
 
     @staticmethod
     def __parse_labels(labels_list: list) -> list:
-        labels = []
-        for label_dict in labels_list:
-            if isinstance(label_dict.labels, ListConfig):
-                labels.append(label_dict.labels)
-            else:
-                labels.append([label_dict.labels])
-        return labels
+        container_labels = []
+        for label_dict in OmegaConf.to_object(labels_list):
+            labels = label_dict.get('labels')
+            if labels is not None:
+                if isinstance(labels, list):
+                    container_labels.append(labels)
+                else:
+                    container_labels.append([labels])
+        return container_labels
 
     @staticmethod
     def __parse_queue_config(queue_config: dict):

tests/config/test_parser.py

@@ -1,6 +1,7 @@
 import os
 
 import pytest
+from omegaconf import ListConfig
 
 from src.pipeline_watchdog.config import WatchConfig
 from src.pipeline_watchdog.config.parser import ConfigParser
@@ -14,6 +15,27 @@ def test_parse(config_file_path, watch_config):
     assert len(config.watch_configs) == 2
     assert config.watch_configs[0] == watch_config
 
+    # check that OmegaConf types are properly converted
+    assert not any(
+        isinstance(labels, ListConfig)
+        for container_labels in [
+            config.watch_configs[0].queue.container_labels,
+            config.watch_configs[0].egress.container_labels,
+            config.watch_configs[0].ingress.container_labels,
+        ]
+        for labels in container_labels
+    )
+    assert all(
+        isinstance(label, str)
+        for container_labels in [
+            config.watch_configs[0].queue.container_labels,
+            config.watch_configs[0].egress.container_labels,
+            config.watch_configs[0].ingress.container_labels,
+        ]
+        for labels in container_labels
+        for label in labels
+    )
+
     # check optional fields
     assert config.watch_configs[1] == WatchConfig(
         buffer='buffer2:8002', queue=None, egress=None, ingress=None
@@ -34,3 +56,11 @@ def test_parse_invalid(invalid_config_file_path):
         match='Field ".*" must be specified in the watch config.',
     ):
         ConfigParser(invalid_config_file_path).parse()
+
+
+def test_parse_empty_labels(invalid_config_with_empty_labels):
+    with pytest.raises(
+        ValueError,
+        match='Container labels cannot be empty.',
+    ):
+        ConfigParser(invalid_config_with_empty_labels).parse()

tests/conftest.py

@@ -88,10 +88,7 @@ def config_file_path():
     )
 )
 def empty_config_file_path(request, tmpdir):
-    config_file = tmpdir.join('empty_config.txt')
-    config_file.write(request.param)
-
-    return str(config_file)
+    return create_tmp_config_file(tmpdir, request.param)
 
 
 @pytest.fixture(
@@ -103,9 +100,61 @@ def empty_config_file_path(request, tmpdir):
     )
 )
 def invalid_config_file_path(request, tmpdir):
+    return create_tmp_config_file(tmpdir, request.param)
+
+
+@pytest.fixture(
+    params=(
+        {
+            'watch': [
+                {
+                    'buffer': 'buffer1:8000',
+                    'queue': {
+                        'action': 'restart',
+                        'length': 18,
+                        'cooldown': '60s',
+                        'polling_interval': '10s',
+                        'container': [],
+                    },
+                }
+            ]
+        },
+        {
+            'watch': [
+                {
+                    'buffer': 'buffer1:8000',
+                    'egress': {
+                        'action': 'restart',
+                        'idle': '100s',
+                        'cooldown': '60s',
+                        'container': [],
+                    },
+                }
+            ]
+        },
+        {
+            'watch': [
+                {
+                    'buffer': 'buffer1:8000',
+                    'ingress': {
+                        'action': 'restart',
+                        'idle': '100s',
+                        'cooldown': '60s',
+                        'container': [],
+                    },
+                }
+            ]
+        },
+    )
+)
+def invalid_config_with_empty_labels(request, tmpdir):
+    return create_tmp_config_file(tmpdir, request.param)
+
+
+def create_tmp_config_file(tmpdir, config):
     config_file = tmpdir.join('empty_config.txt')
 
-    yaml_str = yaml.dump(request.param)
+    yaml_str = yaml.dump(config)
     config_file.write(yaml_str)
 
     return str(config_file)

tests/test_config.yml

@@ -15,7 +15,7 @@ watch:
       polling_interval: ${oc.env:POLLING_INTERVAL}
       container:
         - labels: egress-label=egress-value
-        - labels: some-label
+        - labels: [some-label]
     ingress:
       action: restart
       cooldown: 30s