Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion src/configuration.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from dataclasses import dataclass, field
from enum import Enum

from keboola.component.exceptions import UserException


# the encrypted keys (prefixed with # in Keboola) have to be prefixed with "encrypted_" here
def encrypted_keys(key: str) -> str:
Expand Down Expand Up @@ -50,8 +52,15 @@ class GitConfiguration:
@dataclass
class Configuration:
source: SourceEnum = SourceEnum.CODE
user_properties: dict[str, object] = field(default_factory=dict)
user_properties: dict[str, object] | list = field(default_factory=dict)
venv: VenvEnum = VenvEnum.BASE
packages: list[str] = field(default_factory=list)
code: str = ""
git: GitConfiguration = field(default_factory=GitConfiguration)

def __post_init__(self):
if isinstance(self.user_properties, list):
if len(self.user_properties) == 0:
self.user_properties = {}
else:
raise UserException("Invalid user_properties: non-empty list not supported")
41 changes: 41 additions & 0 deletions tests/test_component.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@

import mock
from freezegun import freeze_time
from keboola.component.exceptions import UserException

from component import Component
from configuration import Configuration


class TestComponent(unittest.TestCase):
Expand All @@ -19,6 +21,45 @@ def test_run_no_cfg_fails(self):
comp.run()


class TestConfigurationUserProperties(unittest.TestCase):
"""Test cases for user_properties handling in Configuration dataclass.

These tests verify the fix for the 'eternal KBC bug' where the Keboola platform
converts empty JSON objects {} to empty arrays [] in configuration parameters.
"""

def test_empty_list_converted_to_empty_dict(self):
"""Empty list [] should be converted to empty dict {} via __post_init__."""
config = Configuration(user_properties=[])
self.assertEqual(config.user_properties, {})
self.assertIsInstance(config.user_properties, dict)

def test_non_empty_list_raises_user_exception(self):
"""Non-empty list should raise UserException."""
with self.assertRaises(UserException) as context:
Configuration(user_properties=["item1", "item2"])
self.assertIn("non-empty list not supported", str(context.exception))

def test_dict_unchanged(self):
"""Normal dict input should remain unchanged."""
test_dict = {"key1": "value1", "key2": 123}
config = Configuration(user_properties=test_dict)
self.assertEqual(config.user_properties, test_dict)
self.assertIsInstance(config.user_properties, dict)

def test_empty_dict_unchanged(self):
"""Empty dict input should remain unchanged."""
config = Configuration(user_properties={})
self.assertEqual(config.user_properties, {})
self.assertIsInstance(config.user_properties, dict)

def test_default_user_properties_is_empty_dict(self):
"""Default user_properties should be an empty dict."""
config = Configuration()
self.assertEqual(config.user_properties, {})
self.assertIsInstance(config.user_properties, dict)


if __name__ == "__main__":
# import sys;sys.argv = ['', 'Test.testName']
unittest.main()