# (c) cavaliba.com - test - test_pipeline.py from django.test import TestCase 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, ) 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")