Introduce pydantic[1] to validate Oxygen handler result value

[1] https://docs.pydantic.dev/2.0/

Author: Tattoo

requirements.txt

@@ -1,6 +1,7 @@
 robotframework>=3.0.4
 junitparser==2.0
 PyYAML>=3.13
+pydantic>=2.4.2
 
 ### Dev
 mock>=2.0.0

setup.py

@@ -34,7 +34,8 @@
       install_requires=[
            'robotframework>=3.0.4',
            'junitparser==2.0',
-           'PyYAML>=3.13'
+           'PyYAML>=3.13',
+           'pydantic>=2.4.2'
       ],
       packages=find_packages(SRC),
       package_dir={'': 'src'},

src/oxygen/errors.py

@@ -32,3 +32,7 @@ class MismatchArgumentException(Exception):
 
 class InvalidConfigurationException(Exception):
     pass
+
+
+class InvalidOxygenResultException(Exception):
+    pass

src/oxygen/oxygen_handler_result.py

@@ -0,0 +1,71 @@
+''' IMPORTANT
+
+OxygenKeywordDict is defined like this since key `pass` is reserved
+word in Python, and thus raises SyntaxError if defined like a class.
+However, in the functional style you cannot refer to the TypedDict itself
+recursively, like you can with with class style. Oh bother.
+
+See more:
+ - https://docs.python.org/3/library/typing.html?highlight=typeddict#typing.TypedDict
+ - https://stackoverflow.com/a/72460065
+'''
+
+import functools
+
+from typing import List
+# TODO FIXME: Python 3.10 requires these to be imported from here
+# Python 3.10 EOL is in 2026
+from typing_extensions import TypedDict, Required
+
+from pydantic import TypeAdapter, ValidationError
+
+from .errors import InvalidOxygenResultException
+
+
+_Pass = TypedDict('_Pass', { 'pass': Required[bool], 'name': Required[str] })
+# define required fields in this one above
+class OxygenKeywordDict(_Pass, total=False):
+    elapsed:  float  # milliseconds
+    tags:     List[str]
+    messages: List[str]
+    teardown: 'OxygenKeywordDict'  # in RF, keywords do not have setup kw; just put it as first kw in `keywords`
+    keywords: List['OxygenKeywordDict']
+
+
+class OxygenTestCaseDict(TypedDict, total=False):
+    name:     Required[str]
+    keywords: Required[List[OxygenKeywordDict]]
+    tags:     List[str]
+    setup:    OxygenKeywordDict
+    teardown: OxygenKeywordDict
+
+
+class OxygenSuiteDict(TypedDict, total=False):
+    name:     Required[str]
+    tags:     List[str]
+    setup:    OxygenKeywordDict
+    teardown: OxygenKeywordDict
+    suites:   List['OxygenSuiteDict']
+    tests:    List[OxygenTestCaseDict]
+
+
+def _change_validationerror_to_oxygenexception(func):
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except ValidationError as e:
+            raise InvalidOxygenResultException(e)
+    return wrapper
+
+@_change_validationerror_to_oxygenexception
+def validate_oxygen_suite(oxygen_result_dict):
+    return TypeAdapter(OxygenSuiteDict).validate_python(oxygen_result_dict)
+
+@_change_validationerror_to_oxygenexception
+def validate_oxygen_test_case(oxygen_test_case_dict):
+    return TypeAdapter(OxygenTestCaseDict).validate_python(oxygen_test_case_dict)
+
+@_change_validationerror_to_oxygenexception
+def validate_oxygen_keyword(oxygen_kw_dict):
+    return TypeAdapter(OxygenKeywordDict).validate_python(oxygen_kw_dict)

tasks.py

@@ -37,7 +37,7 @@ def install(context, package=None):
                   'multiple times to select several targets.'
       })
 def utest(context, test=None):
-    run(f'pytest {" ".join(test) if test else UNIT_TESTS} -q --disable-warnings',
+    run(f'pytest {" -k".join(test) if test else UNIT_TESTS} -q --disable-warnings',
         env={'PYTHONPATH': str(SRCPATH)},
         pty=(not system() == 'Windows'))
 

tests/utest/helpers.py

@@ -8,6 +8,8 @@
 from robot.api import ExecutionResult
 from yaml import FullLoader, load
 
+from oxygen.oxygen_handler_result import OxygenKeywordDict, OxygenTestCaseDict
+
 TEST_CONFIG = '''
 oxygen.junit:
   handler: JUnitHandler
@@ -55,6 +57,24 @@ def example_robot_output():
     output = RESOURCES_PATH / 'example_robot_output.xml'
     return ExecutionResult(output)
 
+MINIMAL_KEYWORD_DICT = { 'name': 'someKeyword', 'pass': True }
+MINIMAL_TC_DICT = { 'name': 'Minimal TC', 'keywords': [MINIMAL_KEYWORD_DICT] }
+
+class _ListSubclass(list):
+    '''Used in test cases'''
+    pass
+
+
+class _KwSubclass(OxygenKeywordDict):
+    '''Used in test cases'''
+    pass
+
+
+class _TCSubclass(OxygenTestCaseDict):
+    '''Used in test cases'''
+    pass
+
+
 GATLING_EXPECTED_OUTPUT = {'name': 'Gatling Scenario',
  'setup': [],
  'suites': [],

tests/utest/oxygen_handler_result/__init__.py

@@ -0,0 +1,45 @@
+from ..helpers import (MINIMAL_KEYWORD_DICT,
+                       _ListSubclass,
+                       _KwSubclass)
+
+class SharedTestsForName(object):
+    def shared_test_for_name(self):
+        class StrSubclass(str):
+            pass
+        valid_inherited = StrSubclass('someKeyword')
+        this_is_not_None = StrSubclass(None)
+
+        self.valid_inputs_for('name',
+                              '',
+                              'someKeyword',
+                              b'someKeyword',
+                              valid_inherited,
+                              this_is_not_None)
+
+        self.invalid_inputs_for('name', None)
+
+
+class SharedTestsForTags(object):
+    def shared_test_for_tags(self):
+        self.valid_inputs_for('tags',
+                              [],
+                              ['some-tag', 'another-tag'],
+                              _ListSubclass())
+
+        invalid_inherited = _ListSubclass()
+        invalid_inherited.append(123)
+
+        self.invalid_inputs_for('tags', [123], None, {'foo': 'bar'}, object())
+
+
+class SharedTestsForKeywordField(object):
+    def shared_test_for_keyword_field(self, attribute):
+        valid_inherited = _KwSubclass(**MINIMAL_KEYWORD_DICT)
+
+        self.valid_inputs_for(attribute,
+                              MINIMAL_KEYWORD_DICT,
+                              valid_inherited,
+                              {**MINIMAL_KEYWORD_DICT,
+                               'something_random': 'will-be-ignored'})
+
+        self.invalid_inputs_for(attribute, None, {})

tests/utest/oxygen_handler_result/shared_tests.py

@@ -0,0 +1,149 @@
+from unittest import TestCase
+
+from oxygen.errors import InvalidOxygenResultException
+from oxygen.oxygen_handler_result import (validate_oxygen_keyword,
+                                          OxygenKeywordDict)
+
+from ..helpers import (MINIMAL_KEYWORD_DICT,
+                       _ListSubclass,
+                       _KwSubclass)
+from .shared_tests import (SharedTestsForKeywordField,
+                           SharedTestsForName,
+                           SharedTestsForTags)
+
+
+class TestOxygenKeywordDict(TestCase,
+                            SharedTestsForName,
+                            SharedTestsForTags,
+                            SharedTestsForKeywordField):
+    def setUp(self):
+        self.minimal = MINIMAL_KEYWORD_DICT
+
+    def test_validate_oxygen_keyword_validates_correctly(self):
+        with self.assertRaises(InvalidOxygenResultException):
+            validate_oxygen_keyword({})
+
+    def test_validate_oxygen_keyword_with_minimal_valid(self):
+        minimal1 = { 'name': 'somename', 'pass': True }
+        minimal2 = { 'name': 'somename', 'pass': False }
+
+        self.assertEqual(validate_oxygen_keyword(minimal1), minimal1)
+        self.assertEqual(validate_oxygen_keyword(minimal2), minimal2)
+
+    def valid_inputs_for(self, attribute, *valid_inputs):
+        for valid_input in valid_inputs:
+            self.assertTrue(validate_oxygen_keyword({**self.minimal,
+                                                     attribute: valid_input}))
+
+    def invalid_inputs_for(self, attribute, *invalid_inputs):
+        for invalid_input in invalid_inputs:
+            with self.assertRaises(InvalidOxygenResultException):
+                validate_oxygen_keyword({**self.minimal,
+                                         attribute: invalid_input})
+
+    def test_validate_oxygen_keyword_validates_name(self):
+        self.shared_test_for_name()
+
+    def test_validate_oxygen_keyword_validates_pass(self):
+        '''
+        Due note that boolean cannot be subclassed in Python:
+        https://mail.python.org/pipermail/python-dev/2002-March/020822.html
+        '''
+        self.valid_inputs_for('pass', True, False, 0, 1, 0.0, 1.0)
+        self.invalid_inputs_for('pass', [], {}, None, object(), -999, -99.9)
+
+    def test_validate_oxygen_keyword_validates_tags(self):
+        self.shared_test_for_tags()
+
+    def test_validate_oxygen_keyword_validates_elapsed(self):
+        class FloatSubclass(float):
+            pass
+
+        self.valid_inputs_for('elapsed',
+                              123.4,
+                              -123.0,
+                              '123.4',
+                              '-999.999',
+                              123,
+                              FloatSubclass())
+
+        self.invalid_inputs_for('elapsed', '', None, object())
+
+    def test_validate_oxygen_keyword_validates_messages(self):
+        valid_inherited = _ListSubclass()
+        valid_inherited.append('message')
+
+        self.valid_inputs_for('messages',
+                              [],
+                              ['message'],
+                              _ListSubclass(),
+                              valid_inherited)
+
+        invalid_inherited = _ListSubclass()
+        invalid_inherited.append('message')
+        invalid_inherited.append(123)
+
+        self.invalid_inputs_for('messages',
+                                'some,messages',
+                                None,
+                                invalid_inherited)
+
+    def test_validate_oxygen_keyword_validates_teardown(self):
+        self.shared_test_for_keyword_field('teardown')
+
+    def test_validate_oxygen_keyword_validates_keywords(self):
+        valid_inherited = _ListSubclass()
+        valid_inherited.append(_KwSubclass(**self.minimal))
+
+        self.valid_inputs_for('keywords',
+                              [],
+                              [self.minimal, {**self.minimal,
+                                              'something_random': 'will-be-ignored'}],
+                              _ListSubclass(),  # empty inherited list
+                              valid_inherited)
+
+        invalid_inherited = _ListSubclass()
+        invalid_inherited.append(_KwSubclass(**self.minimal))
+        invalid_inherited.append(123)
+        self.invalid_inputs_for('keywords', None, invalid_inherited)
+
+    def test_validate_oxygen_keyword_with_maximal_valid(self):
+        expected = {
+            'name': 'keyword',
+            'pass': True,
+            'tags': ['some-tag'],
+            'messages': ['some message'],
+            'teardown': {
+                'name': 'teardownKeyword',
+                'pass': True,
+                'tags': ['teardown-kw'],
+                'messages': ['Teardown passed'],
+                'keywords': []
+            },
+            'keywords': [{
+                'name': 'subKeyword',
+                'pass': False,
+                # tags missing intentionally
+                'messages': ['This particular kw failed'],
+                'teardown': {
+                    'name': 'anotherTeardownKw',
+                    'pass': True,
+                    'tags': ['teardown-kw'],
+                    'messages': ['message from anotherTeardownKw'],
+                    # teardown missing intentionally
+                    'keywords': []
+                },
+                'keywords': [{
+                    'name': 'subsubKeyword',
+                    'pass': True,
+                }]
+            },{
+                'name': 'anotherSubKeyword',
+                'pass': True,
+                'tags': [],
+                'messages': [],
+                'keywords': []
+            }]
+        }
+
+        self.assertEqual(validate_oxygen_keyword(expected), expected)