add value-shape field classification and auto-apply
Field roles are now inferred from observed values instead of names, with src/dst disambiguation, value-pattern detector hints, an Ollama advisor scoped to ambiguous fields only, confidence-gated auto-apply with manual-field protection, and drift flagging for review
This commit is contained in:
88
tests/test_field_shapes.py
Normal file
88
tests/test_field_shapes.py
Normal file
@@ -0,0 +1,88 @@
|
||||
import unittest
|
||||
|
||||
from fgai.field_shapes import classify_field_shape, parse_iso_timestamp, private_ip_ratio
|
||||
|
||||
|
||||
class FieldShapesTests(unittest.TestCase):
|
||||
def test_classifies_ip_shape_without_name_hint(self):
|
||||
values = [f"203.0.113.{n}" for n in range(1, 12)]
|
||||
result = classify_field_shape(values, field_name="x7_field_a")
|
||||
self.assertEqual(result.primary_shape, "ip")
|
||||
self.assertGreater(result.confidence, 60)
|
||||
|
||||
def test_classifies_timestamp_shape_iso(self):
|
||||
values = [f"2026-09-0{n}T12:00:00Z" for n in range(1, 8)]
|
||||
result = classify_field_shape(values, field_name="whatever")
|
||||
self.assertEqual(result.primary_shape, "timestamp")
|
||||
|
||||
def test_classifies_timestamp_shape_epoch(self):
|
||||
values = [str(1757000000 + n) for n in range(10)]
|
||||
result = classify_field_shape(values, field_name="whatever")
|
||||
self.assertEqual(result.primary_shape, "timestamp")
|
||||
|
||||
def test_classifies_port_shape_with_name_hint(self):
|
||||
values = ["80", "443", "22", "3389", "80", "443", "80"]
|
||||
without_name = classify_field_shape(values)
|
||||
with_name = classify_field_shape(values, field_name="dst_port")
|
||||
self.assertEqual(with_name.primary_shape, "port")
|
||||
self.assertGreaterEqual(with_name.confidence, without_name.scores.get("port", 0))
|
||||
|
||||
def test_small_numeric_counter_prefers_numeric_over_port(self):
|
||||
values = ["1", "2", "3", "4", "5", "2", "3"]
|
||||
result = classify_field_shape(values, field_name="retry_count")
|
||||
self.assertEqual(result.primary_shape, "numeric")
|
||||
|
||||
def test_classifies_categorical_low_cardinality(self):
|
||||
values = ["allow", "deny", "allow", "reset", "allow", "deny"]
|
||||
result = classify_field_shape(values, field_name="whatever")
|
||||
self.assertEqual(result.primary_shape, "categorical_low_cardinality")
|
||||
|
||||
def test_classifies_free_text(self):
|
||||
values = [
|
||||
"user bob logged in from workstation A with elevated token set",
|
||||
"connection reset by peer during TLS handshake on segment 4",
|
||||
"unexpected EOF while parsing payload chunk 91 from upstream",
|
||||
"cache miss for key group-7 falling back to origin lookup path",
|
||||
]
|
||||
result = classify_field_shape(values, field_name="message")
|
||||
self.assertEqual(result.primary_shape, "free_text")
|
||||
|
||||
def test_name_bonus_does_not_win_without_shape_evidence(self):
|
||||
# Free-text values named "srcip" should not be classified as ip just because of the name.
|
||||
values = [
|
||||
"user bob logged in from workstation A with elevated token set",
|
||||
"connection reset by peer during TLS handshake on segment 4",
|
||||
"unexpected EOF while parsing payload chunk 91 from upstream",
|
||||
]
|
||||
result = classify_field_shape(values, field_name="srcip")
|
||||
self.assertNotEqual(result.primary_shape, "ip")
|
||||
|
||||
def test_name_bonus_is_bounded(self):
|
||||
values = [f"203.0.113.{n}" for n in range(1, 5)]
|
||||
without_name = classify_field_shape(values)
|
||||
with_name = classify_field_shape(values, field_name="srcip")
|
||||
self.assertLessEqual(with_name.confidence - without_name.confidence, 15)
|
||||
|
||||
def test_empty_values_return_zero_confidence(self):
|
||||
result = classify_field_shape([], field_name="whatever")
|
||||
self.assertEqual(result.confidence, 0)
|
||||
|
||||
def test_parse_iso_timestamp(self):
|
||||
self.assertIsNotNone(parse_iso_timestamp("2026-09-07T12:00:00Z"))
|
||||
self.assertIsNone(parse_iso_timestamp("not-a-timestamp"))
|
||||
|
||||
def test_private_ip_ratio_all_private(self):
|
||||
self.assertEqual(private_ip_ratio(["10.0.0.1", "10.0.0.2", "192.168.1.1"]), 1.0)
|
||||
|
||||
def test_private_ip_ratio_all_public(self):
|
||||
self.assertEqual(private_ip_ratio(["8.8.8.8", "1.1.1.1"]), 0.0)
|
||||
|
||||
def test_private_ip_ratio_mixed(self):
|
||||
self.assertAlmostEqual(private_ip_ratio(["10.0.0.1", "8.8.8.8"]), 0.5)
|
||||
|
||||
def test_private_ip_ratio_ignores_non_ip_values(self):
|
||||
self.assertEqual(private_ip_ratio(["not-an-ip", "also-not"]), 0.0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user