# (c) cavaliba.com - test - test_pipeline.py import app_home.cache as cache from app_data.pipeline import ( condition_empty, condition_equal, condition_exists, condition_ge, condition_gt, condition_is_boolean, condition_is_date, condition_is_datetime, condition_is_false, condition_is_float, condition_is_int, condition_is_ip4, condition_is_subnet, condition_is_time, condition_is_true, condition_le, condition_lt, condition_not_empty, condition_not_equal, task_align_subnet4, task_exit, task_field_append, task_field_copy, task_field_date_now, task_field_datetime, task_field_datetime_now, task_field_delete, task_field_join, task_field_keep, task_field_lower, task_field_md5, task_field_merge, task_field_noop, task_field_nospace, task_field_prepend, task_field_regexp_sub, task_field_rename, task_field_set, task_field_time_now, task_field_tofloat, task_field_toint, task_field_tostring, task_field_truncate, task_field_upper, task_field_uuid, ) from django.test import TestCase class PipelinePrimitive(TestCase): def setUp(self): cache.clear() def test_task_field_noop(self): datadict = {"a":"1234"} taskopt = ["a","-suffix"] task_field_noop(datadict, taskopt) self.assertEqual(len(datadict),1) self.assertEqual(datadict["a"], "1234") def test_task_field_toint(self): datadict = {"a":"1234"} taskopt = ["a"] task_field_toint(datadict, taskopt) self.assertEqual(len(datadict),1) self.assertEqual(datadict["a"], 1234) datadict = {"a":"1234a"} taskopt = ["a"] task_field_toint(datadict, taskopt) self.assertEqual(len(datadict),1) self.assertIsNone(datadict["a"]) def test_task_field_tofloat(self): datadict = {"a":"1234"} taskopt = ["a"] task_field_tofloat(datadict, taskopt) self.assertEqual(len(datadict),1) self.assertEqual(datadict["a"], 1234) datadict = {"a":"1234.5"} taskopt = ["a"] task_field_tofloat(datadict, taskopt) self.assertEqual(len(datadict),1) self.assertEqual(datadict["a"], 1234.5) datadict = {"a":"1234a"} taskopt = ["a"] task_field_tofloat(datadict, taskopt) self.assertEqual(len(datadict),1) self.assertIsNone(datadict["a"]) def test_task_field_append(self): datadict = {"a":"1234"} taskopt = ["a","-suffix"] task_field_append(datadict, taskopt) self.assertEqual(datadict["a"], "1234-suffix") def test_task_field_append_int(self): datadict = {"a":1234} taskopt = ["a","-suffix"] task_field_append(datadict, taskopt) self.assertEqual(datadict["a"], "1234-suffix") datadict = {"a":1234} taskopt = ["a",5678] task_field_append(datadict, taskopt) self.assertEqual(datadict["a"], "12345678") def test_task_field_append_ko(self): datadict = {"a":"1234"} taskopt = ["b","-suffix"] task_field_append(datadict, taskopt) #self.assertEqual(datadict["b"], "-suffix") self.assertNotIn("b", datadict) datadict = {"a":"1234"} taskopt = ["b"] task_field_append(datadict, taskopt) #self.assertEqual(datadict["b"], "-suffix") self.assertNotIn("b", datadict) def test_task_field_prepend(self): datadict = {"a":"1234"} taskopt = ["a","prefix-"] task_field_prepend(datadict, taskopt) self.assertEqual(datadict["a"], "prefix-1234") def test_task_field_prepend_int(self): datadict = {"a":1234} taskopt = ["a","prefix-"] task_field_prepend(datadict, taskopt) self.assertEqual(datadict["a"], "prefix-1234") def test_task_field_prepend_ko(self): datadict = {"a":"1234"} taskopt = ["b","prefix-"] task_field_prepend(datadict, taskopt) #self.assertEqual(datadict["b"], "-suffix") self.assertNotIn("b", datadict) datadict = {"a":"1234"} taskopt = ["b"] task_field_prepend(datadict, taskopt) self.assertNotIn("b", datadict) def test_task_field_md5(self): datadict = {"a":"1234"} taskopt = ["res","a"] task_field_md5(datadict, taskopt) self.assertEqual(datadict["res"], "81dc9bdb52d04dc20036dbd8313ed055") datadict = {"a":"12","b":"34"} taskopt = ["res","a","b"] task_field_md5(datadict, taskopt) self.assertEqual(datadict["res"], "81dc9bdb52d04dc20036dbd8313ed055") datadict = {"a":"12","b":34} taskopt = ["res","a","b"] task_field_md5(datadict, taskopt) self.assertEqual(datadict["res"], "81dc9bdb52d04dc20036dbd8313ed055") datadict = {"a":"12","b":"34"} taskopt = ["res","c"] task_field_md5(datadict, taskopt) self.assertNotIn("res", datadict) datadict = {} taskopt = ["res","c"] task_field_md5(datadict, taskopt) self.assertNotIn("res", datadict) datadict = {"a":"12"} taskopt = ["","a"] task_field_md5(datadict, taskopt) self.assertNotIn("res", datadict) def test_task_field_tostring(self): # Test integer to string datadict = {"a": 5} taskopt = ["a"] task_field_tostring(datadict, taskopt) self.assertEqual(len(datadict), 1) self.assertEqual(datadict["a"], '5') self.assertIsInstance(datadict["a"], str) # Test float to string datadict = {"a": 6.5} taskopt = ["a"] task_field_tostring(datadict, taskopt) self.assertEqual(len(datadict), 1) self.assertEqual(datadict["a"], '6.5') self.assertIsInstance(datadict["a"], str) # Test string remains string datadict = {"a": 'test'} taskopt = ["a"] task_field_tostring(datadict, taskopt) self.assertEqual(len(datadict), 1) self.assertEqual(datadict["a"], 'test') # Test boolean to string datadict = {"a": True} taskopt = ["a"] task_field_tostring(datadict, taskopt) self.assertEqual(len(datadict), 1) self.assertEqual(datadict["a"], 'True') self.assertIsInstance(datadict["a"], str) # Test False boolean to string datadict = {"a": False} taskopt = ["a"] task_field_tostring(datadict, taskopt) self.assertEqual(len(datadict), 1) self.assertEqual(datadict["a"], 'False') # Test None value datadict = {"a": None} taskopt = ["a"] task_field_tostring(datadict, taskopt) self.assertEqual(len(datadict), 1) self.assertEqual(datadict["a"], 'None') # Test multiple fields datadict = {"a": 123, "b": 45.6, "c": True} taskopt = ["a", "b", "c"] task_field_tostring(datadict, taskopt) self.assertEqual(datadict["a"], '123') self.assertEqual(datadict["b"], '45.6') self.assertEqual(datadict["c"], 'True') # Test missing field (should set to None based on implementation) datadict = {"a": 'test'} taskopt = ["b"] task_field_tostring(datadict, taskopt) # Based on the except clause in task_field_tostring, missing key -> None self.assertIn("b", datadict) self.assertIsNone(datadict["b"]) def test_task_field_delete(self): # Test delete single field datadict = {"a": 5, "b": 10} taskopt = ["a"] task_field_delete(datadict, taskopt) self.assertNotIn("a", datadict) self.assertIn("b", datadict) self.assertEqual(datadict["b"], 10) # Test delete multiple fields datadict = {"a": 5, "b": 10, "c": 15, "d": 20, "e": 25} taskopt = ["a", "b"] task_field_delete(datadict, taskopt) self.assertNotIn("a", datadict) self.assertNotIn("b", datadict) self.assertIn("c", datadict) self.assertIn("d", datadict) self.assertIn("e", datadict) self.assertEqual(len(datadict), 3) # Test delete with multiple fields in taskopt (as in original test) datadict = {"a": 5, "b": 5, "c": 5, "d": 5, "e": 5} taskopt = ["a", "b", "c", "d"] task_field_delete(datadict, taskopt) self.assertNotIn("a", datadict) self.assertNotIn("b", datadict) self.assertNotIn("c", datadict) self.assertNotIn("d", datadict) self.assertIn("e", datadict) self.assertEqual(datadict["e"], 5) # Test delete non-existent field (should not raise error) datadict = {"a": 5} taskopt = ["b"] task_field_delete(datadict, taskopt) self.assertIn("a", datadict) self.assertNotIn("b", datadict) self.assertEqual(len(datadict), 1) # Test delete from empty dict datadict = {} taskopt = ["a"] task_field_delete(datadict, taskopt) self.assertEqual(len(datadict), 0) # Test delete all fields datadict = {"a": 1, "b": 2} taskopt = ["a", "b"] task_field_delete(datadict, taskopt) self.assertEqual(len(datadict), 0) def test_task_field_copy(self): # Test basic copy datadict = {"a": 5} taskopt = ["a", "b"] task_field_copy(datadict, taskopt) self.assertIn("a", datadict) self.assertIn("b", datadict) self.assertEqual(datadict["a"], 5) self.assertEqual(datadict["b"], 5) # Test copy with string value datadict = {"a": "test"} taskopt = ["a", "b"] task_field_copy(datadict, taskopt) self.assertEqual(datadict["b"], "test") self.assertEqual(datadict["a"], "test") # Test copy with various types datadict = {"x": 123, "y": "hello", "z": True} taskopt = ["x", "x_copy"] task_field_copy(datadict, taskopt) self.assertEqual(datadict["x_copy"], 123) taskopt = ["y", "y_copy"] task_field_copy(datadict, taskopt) self.assertEqual(datadict["y_copy"], "hello") # Test copy non-existent field (should fail silently) datadict = {"a": 5} taskopt = ["b", "c"] task_field_copy(datadict, taskopt) self.assertNotIn("c", datadict) # Test overwrite existing field datadict = {"a": 5, "b": 10} taskopt = ["a", "b"] task_field_copy(datadict, taskopt) self.assertEqual(datadict["b"], 5) def test_task_field_rename(self): # Test basic rename datadict = {"a": 5} taskopt = ["a", "b"] task_field_rename(datadict, taskopt) self.assertNotIn("a", datadict) self.assertIn("b", datadict) self.assertEqual(datadict["b"], 5) # Test rename with string value datadict = {"old_name": "value"} taskopt = ["old_name", "new_name"] task_field_rename(datadict, taskopt) self.assertNotIn("old_name", datadict) self.assertIn("new_name", datadict) self.assertEqual(datadict["new_name"], "value") # Test rename doesn't affect other fields datadict = {"a": 1, "b": 2, "c": 3} taskopt = ["a", "d"] task_field_rename(datadict, taskopt) self.assertNotIn("a", datadict) self.assertIn("b", datadict) self.assertIn("c", datadict) self.assertIn("d", datadict) self.assertEqual(datadict["d"], 1) self.assertEqual(len(datadict), 3) # Test rename non-existent field (should fail silently) datadict = {"a": 5} taskopt = ["b", "c"] task_field_rename(datadict, taskopt) self.assertIn("a", datadict) self.assertNotIn("b", datadict) self.assertNotIn("c", datadict) # Test rename to existing field (overwrites) datadict = {"a": 5, "b": 10} taskopt = ["a", "b"] task_field_rename(datadict, taskopt) self.assertNotIn("a", datadict) self.assertEqual(datadict["b"], 5) def test_task_field_upper(self): # Test basic uppercase datadict = {"a": "test"} taskopt = ["a"] task_field_upper(datadict, taskopt) self.assertEqual(datadict["a"], "TEST") # Test with mixed case datadict = {"a": "TeSt"} taskopt = ["a"] task_field_upper(datadict, taskopt) self.assertEqual(datadict["a"], "TEST") # Test with multiple fields datadict = {"a": "hello", "b": "world"} taskopt = ["a", "b"] task_field_upper(datadict, taskopt) self.assertEqual(datadict["a"], "HELLO") self.assertEqual(datadict["b"], "WORLD") # Test with already uppercase datadict = {"a": "TEST"} taskopt = ["a"] task_field_upper(datadict, taskopt) self.assertEqual(datadict["a"], "TEST") # Test with non-string field (should fail silently) datadict = {"a": 123} taskopt = ["a"] task_field_upper(datadict, taskopt) # Should remain unchanged due to exception handling self.assertEqual(datadict["a"], 123) # Test with non-existent field (should fail silently) datadict = {"a": "test"} taskopt = ["b"] task_field_upper(datadict, taskopt) self.assertNotIn("b", datadict) def test_task_field_lower(self): # Test basic lowercase datadict = {"a": "TEST"} taskopt = ["a"] task_field_lower(datadict, taskopt) self.assertEqual(datadict["a"], "test") # Test with mixed case datadict = {"a": "TeSt"} taskopt = ["a"] task_field_lower(datadict, taskopt) self.assertEqual(datadict["a"], "test") # Test with multiple fields datadict = {"a": "HELLO", "b": "WORLD"} taskopt = ["a", "b"] task_field_lower(datadict, taskopt) self.assertEqual(datadict["a"], "hello") self.assertEqual(datadict["b"], "world") # Test with already lowercase datadict = {"a": "test"} taskopt = ["a"] task_field_lower(datadict, taskopt) self.assertEqual(datadict["a"], "test") # Test with non-string field (should fail silently) datadict = {"a": 123} taskopt = ["a"] task_field_lower(datadict, taskopt) # Should remain unchanged due to exception handling self.assertEqual(datadict["a"], 123) # Test with non-existent field (should fail silently) datadict = {"a": "TEST"} taskopt = ["b"] task_field_lower(datadict, taskopt) self.assertNotIn("b", datadict) def test_task_field_set(self): # Test set empty string datadict = {} taskopt = ["a", ""] task_field_set(datadict, taskopt) self.assertIn("a", datadict) self.assertEqual(datadict["a"], "") # Test set float value datadict = {} taskopt = ["b", 5.0] task_field_set(datadict, taskopt) self.assertEqual(datadict["b"], 5.0) # Test set string value datadict = {} taskopt = ["c", "test"] task_field_set(datadict, taskopt) self.assertEqual(datadict["c"], "test") # Test set with invalid taskopt (missing value, should fail silently) datadict = {} taskopt = ["d"] task_field_set(datadict, taskopt) self.assertNotIn("d", datadict) # Test overwrite existing value datadict = {"a": "old"} taskopt = ["a", "new"] task_field_set(datadict, taskopt) self.assertEqual(datadict["a"], "new") # Test set integer datadict = {} taskopt = ["num", 42] task_field_set(datadict, taskopt) self.assertEqual(datadict["num"], 42) # Test set boolean datadict = {} taskopt = ["flag", True] task_field_set(datadict, taskopt) self.assertEqual(datadict["flag"], True) def test_task_field_merge(self): # Test merge integers datadict = {"a": 5, "b": 6} taskopt = ["a", "b", "c"] task_field_merge(datadict, taskopt) self.assertIn("c", datadict) self.assertEqual(datadict["c"], 11) # Test merge strings datadict = {"a": "unit", "b": "test"} taskopt = ["a", "b", "c"] task_field_merge(datadict, taskopt) self.assertIn("c", datadict) self.assertEqual(datadict["c"], "unittest") # Test merge floats datadict = {"x": 1.5, "y": 2.5} taskopt = ["x", "y", "z"] task_field_merge(datadict, taskopt) self.assertEqual(datadict["z"], 4.0) # Test merge missing field (should fail silently) datadict = {"a": 5} taskopt = ["a", "b", "c"] task_field_merge(datadict, taskopt) self.assertNotIn("c", datadict) # Test overwrite existing result field datadict = {"a": 1, "b": 2, "c": 999} taskopt = ["a", "b", "c"] task_field_merge(datadict, taskopt) self.assertEqual(datadict["c"], 3) def test_task_field_uuid(self): # Test basic UUID generation datadict = {} taskopt = ["uuid"] task_field_uuid(datadict, taskopt) self.assertIn("uuid", datadict) self.assertEqual(len(datadict["uuid"]), 36) self.assertIn("-", datadict["uuid"]) # Test multiple UUID fields datadict = {} taskopt = ["uuid1", "uuid2"] task_field_uuid(datadict, taskopt) self.assertIn("uuid1", datadict) self.assertIn("uuid2", datadict) self.assertEqual(len(datadict["uuid1"]), 36) self.assertEqual(len(datadict["uuid2"]), 36) # UUIDs should be different self.assertNotEqual(datadict["uuid1"], datadict["uuid2"]) # Test UUID format (8-4-4-4-12 characters) datadict = {} taskopt = ["id"] task_field_uuid(datadict, taskopt) parts = datadict["id"].split("-") self.assertEqual(len(parts), 5) self.assertEqual(len(parts[0]), 8) self.assertEqual(len(parts[1]), 4) self.assertEqual(len(parts[2]), 4) self.assertEqual(len(parts[3]), 4) self.assertEqual(len(parts[4]), 12) def test_task_field_datetime(self): # Test date_now datadict = {} taskopt = ["date"] task_field_date_now(datadict, taskopt) self.assertIn("date", datadict) self.assertTrue(datadict["date"].startswith("20")) self.assertEqual(len(datadict["date"]), 10) # YYYY-MM-DD self.assertEqual(datadict["date"][4], "-") self.assertEqual(datadict["date"][7], "-") # Test datetime_now datadict = {} taskopt = ["datetime"] task_field_datetime_now(datadict, taskopt) self.assertIn("datetime", datadict) self.assertTrue(datadict["datetime"].startswith("20")) self.assertEqual(len(datadict["datetime"]), 19) # YYYY-MM-DD HH:MM:SS self.assertEqual(datadict["datetime"][10], " ") # Test time_now datadict = {} taskopt = ["time"] task_field_time_now(datadict, taskopt) self.assertIn("time", datadict) self.assertEqual(len(datadict["time"]), 8) # HH:MM:SS self.assertEqual(datadict["time"][2], ":") self.assertEqual(datadict["time"][5], ":") # Test multiple datetime fields datadict = {} taskopt = ["a"] task_field_date_now(datadict, taskopt) taskopt = ["b"] task_field_datetime_now(datadict, taskopt) taskopt = ["c"] task_field_time_now(datadict, taskopt) self.assertIn("a", datadict) self.assertIn("b", datadict) self.assertIn("c", datadict) def test_task_field_regexp_sub(self): # Test basic substitution datadict = {"a": "This is a test from unittest !"} taskopt = ["a", "test", "QWERTY"] task_field_regexp_sub(datadict, taskopt) self.assertEqual(datadict["a"], "This is a QWERTY from unitQWERTY !") # Test single substitution datadict = {"text": "hello world"} taskopt = ["text", "world", "universe"] task_field_regexp_sub(datadict, taskopt) self.assertEqual(datadict["text"], "hello universe") # Test regex pattern (digits) datadict = {"code": "abc123def456"} taskopt = ["code", r"\d+", "X"] task_field_regexp_sub(datadict, taskopt) self.assertEqual(datadict["code"], "abcXdefX") # Test no match (should remain unchanged) datadict = {"a": "hello"} taskopt = ["a", "xyz", "ABC"] task_field_regexp_sub(datadict, taskopt) self.assertEqual(datadict["a"], "hello") # Test missing field (should fail silently) datadict = {"a": "test"} taskopt = ["b", "test", "replace"] task_field_regexp_sub(datadict, taskopt) self.assertNotIn("b", datadict) # Test remove pattern (replace with empty) datadict = {"a": "test123test"} taskopt = ["a", r"\d+", ""] task_field_regexp_sub(datadict, taskopt) self.assertEqual(datadict["a"], "testtest") def test_task_field_keep(self): # Test keep single field datadict = {"a": 5, "b": 5, "c": 5, "d": 5} taskopt = ["a"] task_field_keep(datadict, taskopt) self.assertIn("a", datadict) self.assertNotIn("b", datadict) self.assertNotIn("c", datadict) self.assertNotIn("d", datadict) self.assertEqual(datadict["a"], 5) self.assertEqual(len(datadict), 1) # Test keep multiple fields datadict = {"a": 1, "b": 2, "c": 3, "d": 4, "e": 5} taskopt = ["a", "c", "e"] task_field_keep(datadict, taskopt) self.assertIn("a", datadict) self.assertNotIn("b", datadict) self.assertIn("c", datadict) self.assertNotIn("d", datadict) self.assertIn("e", datadict) self.assertEqual(len(datadict), 3) # Test keep field that doesn't exist (should keep nothing except it) datadict = {"a": 1, "b": 2} taskopt = ["c"] task_field_keep(datadict, taskopt) self.assertNotIn("a", datadict) self.assertNotIn("b", datadict) self.assertNotIn("c", datadict) self.assertEqual(len(datadict), 0) # Test keep all existing fields datadict = {"x": "test1", "y": "test2"} taskopt = ["x", "y"] task_field_keep(datadict, taskopt) self.assertEqual(len(datadict), 2) self.assertIn("x", datadict) self.assertIn("y", datadict) # Test with classname and keyname (should be preserved automatically) datadict = {"classname": "test", "keyname": "key1", "a": 1, "b": 2} taskopt = ["a"] task_field_keep(datadict, taskopt) # classname and keyname should be preserved even though not in taskopt self.assertIn("classname", datadict) self.assertIn("keyname", datadict) self.assertIn("a", datadict) self.assertNotIn("b", datadict) # Test empty taskopt (keeps only classname/keyname if present) datadict = {"a": 1, "b": 2, "c": 3} taskopt = [] task_field_keep(datadict, taskopt) self.assertEqual(len(datadict), 0) def test_task_field_nospace(self): # Test remove spaces from single word with spaces datadict = {"a": "hello world"} taskopt = ["a"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "helloworld") # Test remove multiple spaces datadict = {"text": "this has many spaces"} taskopt = ["text"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["text"], "thishasmanyspaces") # Test remove leading/trailing spaces datadict = {"a": " trim me "} taskopt = ["a"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "trimme") # Test with tabs and newlines datadict = {"a": "hello\tworld\ntest"} taskopt = ["a"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "helloworldtest") # Test multiple fields datadict = {"a": "hello world", "b": "foo bar"} taskopt = ["a", "b"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "helloworld") self.assertEqual(datadict["b"], "foobar") # Test with no spaces (idempotent) datadict = {"a": "nospaces"} taskopt = ["a"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "nospaces") # Test with non-string field (should fail silently) datadict = {"a": 123} taskopt = ["a"] task_field_nospace(datadict, taskopt) # Should remain unchanged due to exception handling self.assertEqual(datadict["a"], 123) # Test with non-existent field (should fail silently) datadict = {"a": "test"} taskopt = ["b"] task_field_nospace(datadict, taskopt) self.assertNotIn("b", datadict) self.assertEqual(datadict["a"], "test") # Test empty string datadict = {"a": ""} taskopt = ["a"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "") # Test string with only spaces datadict = {"a": " "} taskopt = ["a"] task_field_nospace(datadict, taskopt) self.assertEqual(datadict["a"], "") def test_task_align_subnet4(self): # Test basic alignment datadict = {"subnet": "10.1.1.1/24"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) self.assertEqual(datadict["subnet"], "10.1.1.0/24") # Test already aligned subnet datadict = {"subnet": "10.1.1.0/24"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) self.assertEqual(datadict["subnet"], "10.1.1.0/24") # Test /32 subnet (single IP) datadict = {"ip": "192.168.1.100/32"} taskopt = ["ip"] task_align_subnet4(datadict, taskopt) self.assertEqual(datadict["ip"], "192.168.1.100/32") # Test /25 subnet datadict = {"subnet": "172.16.0.200/25"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) self.assertEqual(datadict["subnet"], "172.16.0.128/25") # Test /8 subnet datadict = {"subnet": "10.100.200.50/8"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) self.assertEqual(datadict["subnet"], "10.0.0.0/8") # Test with missing field (should fail silently) datadict = {"a": "10.1.1.1/24"} taskopt = ["b"] task_align_subnet4(datadict, taskopt) self.assertNotIn("b", datadict) self.assertEqual(datadict["a"], "10.1.1.1/24") # Test with invalid IP/mask (should fail silently) datadict = {"subnet": "invalid"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) # Should remain unchanged due to exception handling self.assertEqual(datadict["subnet"], "invalid") # Test with missing mask (should add /32 mask) datadict = {"subnet": "10.1.1.1"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) # Should be treated as /32 subnet self.assertEqual(datadict["subnet"], "10.1.1.1/32") # Test /16 subnet alignment datadict = {"subnet": "192.168.50.100/16"} taskopt = ["subnet"] task_align_subnet4(datadict, taskopt) self.assertEqual(datadict["subnet"], "192.168.0.0/16") def test_task_field_join(self): # Test basic join with dash separator datadict = {"a": "hello", "b": "world"} taskopt = ["-", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertIn("result", datadict) self.assertEqual(datadict["result"], "hello-world") # Test join with space separator datadict = {"first": "John", "last": "Doe"} taskopt = [" ", "fullname", "first", "last"] task_field_join(datadict, taskopt) self.assertEqual(datadict["fullname"], "John Doe") # Test join with empty separator datadict = {"a": "foo", "b": "bar"} taskopt = ["", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "foobar") # Test join with multiple fields datadict = {"a": "one", "b": "two", "c": "three"} taskopt = [",", "result", "a", "b", "c"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "one,two,three") # Test join with integer fields (should convert to string) datadict = {"a": 10, "b": 20, "c": 30} taskopt = [":", "result", "a", "b", "c"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "10:20:30") # Test join with mixed types datadict = {"a": "hello", "b": 42, "c": 3.14} taskopt = ["|", "result", "a", "b", "c"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "hello|42|3.14") # Test join with missing field (should fail silently) datadict = {"a": "hello"} taskopt = ["-", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertNotIn("result", datadict) # Test join with single field datadict = {"a": "single"} taskopt = ["-", "result", "a"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "single") # Test join overwrites existing destination field datadict = {"a": "new", "b": "value", "result": "old"} taskopt = ["-", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "new-value") # Test join with multi-character separator datadict = {"a": "start", "b": "middle", "c": "end"} taskopt = [" --> ", "result", "a", "b", "c"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "start --> middle --> end") # Test join with special characters in separator datadict = {"a": "part1", "b": "part2"} taskopt = [":::", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "part1:::part2") # Test join with empty string value in field datadict = {"a": "", "b": "value"} taskopt = ["-", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "-value") # Test join with boolean values (should convert to string) datadict = {"a": True, "b": False} taskopt = [",", "result", "a", "b"] task_field_join(datadict, taskopt) self.assertEqual(datadict["result"], "True,False") def test_task_exit(self): datadict = {'a': 'value'} status = task_exit(datadict, []) self.assertEqual(status, 'exit') def test_task_exit_custom_status(self): datadict = {'a': 'value'} status = task_exit(datadict, ['myexit']) self.assertEqual(status, 'myexit') def test_task_field_datetime_now(self): from datetime import datetime datadict = {} task_field_datetime(datadict, ['result', 'now()']) self.assertIn('result', datadict) dt = datetime.fromisoformat(datadict['result']) self.assertIsInstance(dt, datetime) def test_task_field_datetime_from_string(self): datadict = {} task_field_datetime(datadict, ['result', '2025-06-01 10:00:00']) self.assertEqual(datadict['result'], '2025-06-01 10:00:00') def test_task_field_datetime_from_field(self): datadict = {'start': '2025-03-15 08:30:00'} task_field_datetime(datadict, ['result', 'start']) self.assertEqual(datadict['result'], '2025-03-15 08:30:00') def test_task_field_datetime_delta_add_days(self): datadict = {} task_field_datetime(datadict, ['result', '2025-01-01 00:00:00', '+10day']) self.assertEqual(datadict['result'], '2025-01-11 00:00:00') def test_task_field_datetime_delta_sub_hours(self): datadict = {} task_field_datetime(datadict, ['result', '2025-06-15 12:00:00', '-3hour']) self.assertEqual(datadict['result'], '2025-06-15 09:00:00') def test_task_field_datetime_delta_add_months(self): datadict = {} task_field_datetime(datadict, ['result', '2025-01-31 00:00:00', '+1month']) self.assertEqual(datadict['result'], '2025-02-28 00:00:00') def test_task_field_datetime_delta_add_year(self): datadict = {} task_field_datetime(datadict, ['result', '2024-02-29 00:00:00', '+1year']) self.assertEqual(datadict['result'], '2025-02-28 00:00:00') def test_task_field_datetime_delta_from_field(self): datadict = {'delta': '+5day'} task_field_datetime(datadict, ['result', '2025-04-01 00:00:00', 'delta']) self.assertEqual(datadict['result'], '2025-04-06 00:00:00') def test_task_field_datetime_invalid_source(self): datadict = {} task_field_datetime(datadict, ['result', 'not_a_date_and_not_a_field']) self.assertNotIn('result', datadict) def test_task_field_datetime_missing_args(self): datadict = {} task_field_datetime(datadict, ['result']) self.assertNotIn('result', datadict) # ------------------------------------------------------------------ # condition comparators # ------------------------------------------------------------------ def test_condition_ge_int(self): datadict = {'a': '10', 'b': '5'} self.assertTrue(condition_ge(datadict, ['a', 'b'])) self.assertTrue(condition_ge(datadict, ['b', 'b'])) self.assertFalse(condition_ge(datadict, ['b', 'a'])) def test_condition_ge_float(self): datadict = {'a': '3.5', 'b': '3.14'} self.assertTrue(condition_ge(datadict, ['a', 'b'])) self.assertFalse(condition_ge(datadict, ['b', 'a'])) def test_condition_ge_string(self): datadict = {'a': 'banana', 'b': 'apple'} self.assertTrue(condition_ge(datadict, ['a', 'b'])) self.assertFalse(condition_ge(datadict, ['b', 'a'])) def test_condition_gt_int(self): datadict = {'a': '10', 'b': '5'} self.assertTrue(condition_gt(datadict, ['a', 'b'])) self.assertFalse(condition_gt(datadict, ['b', 'a'])) self.assertFalse(condition_gt(datadict, ['b', 'b'])) def test_condition_lt_int(self): datadict = {'a': '3', 'b': '7'} self.assertTrue(condition_lt(datadict, ['a', 'b'])) self.assertFalse(condition_lt(datadict, ['b', 'a'])) self.assertFalse(condition_lt(datadict, ['a', 'a'])) def test_condition_le_int(self): datadict = {'a': '5', 'b': '10'} self.assertTrue(condition_le(datadict, ['a', 'b'])) self.assertTrue(condition_le(datadict, ['a', 'a'])) self.assertFalse(condition_le(datadict, ['b', 'a'])) def test_condition_eq_int(self): datadict = {'a': '42', 'b': '42', 'c': '0'} self.assertTrue(condition_equal(datadict, ['a', 'b'])) self.assertFalse(condition_equal(datadict, ['a', 'c'])) def test_condition_eq_string(self): datadict = {'a': 'hello', 'b': 'hello', 'c': 'world'} self.assertTrue(condition_equal(datadict, ['a', 'b'])) self.assertFalse(condition_equal(datadict, ['a', 'c'])) def test_condition_ne_int(self): datadict = {'a': '1', 'b': '2'} self.assertTrue(condition_not_equal(datadict, ['a', 'b'])) self.assertFalse(condition_not_equal(datadict, ['a', 'a'])) def test_condition_exists_present(self): datadict = {'a': 'hello', 'b': ''} self.assertTrue(condition_exists(datadict, ['a'])) self.assertTrue(condition_exists(datadict, ['b'])) def test_condition_exists_missing(self): datadict = {'a': 'hello'} self.assertFalse(condition_exists(datadict, ['z'])) def test_condition_exists_no_args(self): datadict = {'a': 'hello'} self.assertFalse(condition_exists(datadict, [])) def test_condition_empty_is_empty(self): datadict = {'a': '', 'b': ' '} self.assertTrue(condition_empty(datadict, ['a'])) self.assertTrue(condition_empty(datadict, ['b'])) def test_condition_empty_not_empty(self): datadict = {'a': 'hello'} self.assertFalse(condition_empty(datadict, ['a'])) def test_condition_empty_missing_field(self): datadict = {'a': 'hello'} self.assertFalse(condition_empty(datadict, ['z'])) def test_condition_empty_no_args(self): datadict = {'a': ''} self.assertFalse(condition_empty(datadict, [])) def test_condition_not_empty_has_value(self): datadict = {'a': 'hello'} self.assertTrue(condition_not_empty(datadict, ['a'])) def test_condition_not_empty_is_empty(self): datadict = {'a': '', 'b': ' '} self.assertFalse(condition_not_empty(datadict, ['a'])) self.assertFalse(condition_not_empty(datadict, ['b'])) def test_condition_not_empty_missing_field(self): datadict = {'a': 'hello'} self.assertFalse(condition_not_empty(datadict, ['z'])) def test_condition_not_empty_no_args(self): datadict = {'a': 'hello'} self.assertFalse(condition_not_empty(datadict, [])) def test_condition_is_true_truthy_values(self): for val in ('on', 'On', 'ON', True, 'yes', 'Yes', 'YES', 'True', 'true', 'TRUE', 1, "1"): datadict = {'a': val} self.assertTrue(condition_is_true(datadict, ['a']), msg=f"expected True for {val!r}") def test_condition_is_true_falsy_values(self): for val in ('false', 'False', 'no', '0', 0, '', 'off'): datadict = {'a': val} self.assertFalse(condition_is_true(datadict, ['a']), msg=f"expected False for {val!r}") def test_condition_is_true_missing_field(self): datadict = {'a': 'true'} self.assertFalse(condition_is_true(datadict, ['z'])) def test_condition_is_true_no_args(self): datadict = {'a': 'true'} self.assertFalse(condition_is_true(datadict, [])) def test_condition_is_false_falsy_values(self): for val in ('false', 'False', 'no', '0', 0, '', 'off'): datadict = {'a': val} self.assertTrue(condition_is_false(datadict, ['a']), msg=f"expected True for {val!r}") def test_condition_is_false_truthy_values(self): for val in ('on', 'On', 'ON', True, 'yes', 'Yes', 'YES', 'True', 'true', 'TRUE', 1, "1"): datadict = {'a': val} self.assertFalse(condition_is_false(datadict, ['a']), msg=f"expected False for {val!r}") def test_condition_is_false_missing_field(self): datadict = {'a': 'true'} self.assertFalse(condition_is_false(datadict, ['z'])) def test_condition_is_false_no_args(self): datadict = {'a': 'false'} self.assertFalse(condition_is_false(datadict, [])) def test_condition_is_int_valid(self): for val in ('0', '42', '-7', '1000'): self.assertTrue(condition_is_int({'a': val}, ['a']), msg=f"expected True for {val!r}") def test_condition_is_int_invalid(self): for val in ('3.14', 'abc', '', 'None'): self.assertFalse(condition_is_int({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_int_missing_field(self): self.assertFalse(condition_is_int({'a': '1'}, ['z'])) def test_condition_is_int_no_args(self): self.assertFalse(condition_is_int({'a': '1'}, [])) def test_condition_is_float_valid(self): for val in ('3.14', '0.0', '-1.5', '42', '1e3'): self.assertTrue(condition_is_float({'a': val}, ['a']), msg=f"expected True for {val!r}") def test_condition_is_float_invalid(self): for val in ('abc', '', 'None'): self.assertFalse(condition_is_float({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_float_missing_field(self): self.assertFalse(condition_is_float({'a': '1.0'}, ['z'])) def test_condition_is_float_no_args(self): self.assertFalse(condition_is_float({'a': '1.0'}, [])) def test_condition_is_boolean_valid(self): for val in ('true', 'True', 'TRUE', 'false', 'False', 'FALSE', 'yes', 'Yes', 'YES', 'no', 'No', 'NO', 'on', 'On', 'ON', '1', '0', 1, 0, True, False): self.assertTrue(condition_is_boolean({'a': val}, ['a']), msg=f"expected True for {val!r}") def test_condition_is_boolean_invalid(self): for val in ('maybe', 'ok', '2', 'yep', ''): self.assertFalse(condition_is_boolean({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_boolean_missing_field(self): self.assertFalse(condition_is_boolean({'a': 'true'}, ['z'])) def test_condition_is_boolean_no_args(self): self.assertFalse(condition_is_boolean({'a': 'true'}, [])) def test_condition_is_date_valid(self): self.assertTrue(condition_is_date({'a': '2025-06-15'}, ['a'])) def test_condition_is_date_invalid(self): for val in ('2025-06-15 12:00:00', '15/06/2025', 'abc', ''): self.assertFalse(condition_is_date({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_date_missing_field(self): self.assertFalse(condition_is_date({'a': '2025-06-15'}, ['z'])) def test_condition_is_date_no_args(self): self.assertFalse(condition_is_date({'a': '2025-06-15'}, [])) def test_condition_is_time_valid(self): self.assertTrue(condition_is_time({'a': '14:30:00'}, ['a'])) def test_condition_is_time_invalid(self): for val in ('14:30', '2025-06-15', 'abc', ''): self.assertFalse(condition_is_time({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_time_missing_field(self): self.assertFalse(condition_is_time({'a': '14:30:00'}, ['z'])) def test_condition_is_time_no_args(self): self.assertFalse(condition_is_time({'a': '14:30:00'}, [])) def test_condition_is_datetime_valid(self): self.assertTrue(condition_is_datetime({'a': '2025-06-15 14:30:00'}, ['a'])) def test_condition_is_datetime_invalid(self): for val in ('2025-06-15', '14:30:00', 'abc', ''): self.assertFalse(condition_is_datetime({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_datetime_missing_field(self): self.assertFalse(condition_is_datetime({'a': '2025-06-15 14:30:00'}, ['z'])) def test_condition_is_datetime_no_args(self): self.assertFalse(condition_is_datetime({'a': '2025-06-15 14:30:00'}, [])) def test_condition_is_ip4_valid(self): for val in ('192.168.1.1', '10.0.0.0', '0.0.0.0', '255.255.255.255'): self.assertTrue(condition_is_ip4({'a': val}, ['a']), msg=f"expected True for {val!r}") def test_condition_is_ip4_invalid(self): for val in ('192.168.1.1/24', '256.0.0.1', '192.168.1', 'abc', ''): self.assertFalse(condition_is_ip4({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_ip4_missing_field(self): self.assertFalse(condition_is_ip4({'a': '10.0.0.1'}, ['z'])) def test_condition_is_ip4_no_args(self): self.assertFalse(condition_is_ip4({'a': '10.0.0.1'}, [])) def test_condition_is_subnet_valid(self): for val in ('192.168.1.0/24', '10.0.0.0/8', '0.0.0.0/0', '172.16.0.0/12'): self.assertTrue(condition_is_subnet({'a': val}, ['a']), msg=f"expected True for {val!r}") def test_condition_is_subnet_invalid(self): for val in ('192.168.1.1', '192.168.1.0/33', '192.168.1.0/abc', 'abc', ''): self.assertFalse(condition_is_subnet({'a': val}, ['a']), msg=f"expected False for {val!r}") def test_condition_is_subnet_missing_field(self): self.assertFalse(condition_is_subnet({'a': '10.0.0.0/8'}, ['z'])) def test_condition_is_subnet_no_args(self): self.assertFalse(condition_is_subnet({'a': '10.0.0.0/8'}, [])) def test_condition_compare_int_vs_string_numeric(self): # "9" and "10" should compare as integers, not strings datadict = {'a': '10', 'b': '9'} self.assertTrue(condition_gt(datadict, ['a', 'b'])) def test_condition_compare_missing_field(self): datadict = {'a': '5'} self.assertFalse(condition_ge(datadict, ['a', 'missing'])) self.assertFalse(condition_lt(datadict, ['missing', 'a'])) def test_condition_compare_dates_gt(self): datadict = {'a': '2025-06-01 00:00:00', 'b': '2024-12-31 23:59:59'} self.assertTrue(condition_gt(datadict, ['a', 'b'])) self.assertFalse(condition_gt(datadict, ['b', 'a'])) def test_condition_compare_dates_lt(self): datadict = {'a': '2024-01-01 00:00:00', 'b': '2025-01-01 00:00:00'} self.assertTrue(condition_lt(datadict, ['a', 'b'])) self.assertFalse(condition_lt(datadict, ['b', 'a'])) def test_condition_compare_dates_eq(self): datadict = {'a': '2025-03-15 12:00:00', 'b': '2025-03-15 12:00:00'} self.assertTrue(condition_equal(datadict, ['a', 'b'])) def test_condition_compare_dates_ne(self): datadict = {'a': '2025-03-15 12:00:00', 'b': '2025-03-15 12:00:01'} self.assertTrue(condition_not_equal(datadict, ['a', 'b'])) def test_condition_compare_dates_ge(self): datadict = {'a': '2025-06-01 08:00:00', 'b': '2025-06-01 08:00:00'} self.assertTrue(condition_ge(datadict, ['a', 'b'])) datadict['a'] = '2025-06-01 08:00:01' self.assertTrue(condition_ge(datadict, ['a', 'b'])) self.assertFalse(condition_ge(datadict, ['b', 'a'])) def test_condition_compare_dates_le(self): datadict = {'a': '2025-01-01 00:00:00', 'b': '2025-12-31 23:59:59'} self.assertTrue(condition_le(datadict, ['a', 'b'])) self.assertFalse(condition_le(datadict, ['b', 'a'])) def test_task_field_truncate(self): datadict = {"a": "hello world"} task_field_truncate(datadict, ["a", 5]) self.assertEqual(datadict["a"], "hello") def test_task_field_truncate_exact(self): datadict = {"a": "hello"} task_field_truncate(datadict, ["a", 5]) self.assertEqual(datadict["a"], "hello") def test_task_field_truncate_longer_than_value(self): datadict = {"a": "hi"} task_field_truncate(datadict, ["a", 100]) self.assertEqual(datadict["a"], "hi") def test_task_field_truncate_zero(self): datadict = {"a": "hello"} task_field_truncate(datadict, ["a", 0]) self.assertEqual(datadict["a"], "") def test_task_field_truncate_int_value(self): datadict = {"a": 123456} task_field_truncate(datadict, ["a", 3]) self.assertEqual(datadict["a"], "123") def test_task_field_truncate_missing_field(self): datadict = {"a": "hello"} task_field_truncate(datadict, ["b", 3]) self.assertNotIn("b", datadict) self.assertEqual(datadict["a"], "hello")