caa37de98e
classify_field_shape() received a deduplicated set of observed values with no real event-volume context, so cardinality was always measured against itself — trivially 100%, making any field with more than ~16 distinct values look "high cardinality" no matter how large the real traffic volume was. This misclassified genuinely low-cardinality fields (e.g. username: 82 distinct values across 1,601 events) as free text, which the new shape-veto work then surfaced as false-positive "remove this field" warnings in production. Fixed by passing real event coverage as an explicit total_events parameter; omitting it keeps prior behavior for other callers. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
108 lines
5.2 KiB
Python
108 lines
5.2 KiB
Python
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_deduplicated_sample_alone_misjudges_cardinality_without_total_events(self):
|
|
# sample_values is commonly a deduplicated set (callers cap distinct values
|
|
# remembered per field), so its own length can't stand in for real event
|
|
# volume: 82 distinct plain usernames look "high cardinality" against
|
|
# themselves even though they're a tiny fraction of a much larger stream.
|
|
values = [f"user{n}" for n in range(82)]
|
|
without_total = classify_field_shape(values, field_name="username")
|
|
self.assertEqual(without_total.primary_shape, "free_text")
|
|
|
|
def test_total_events_corrects_cardinality_judgement_for_low_cardinality_field(self):
|
|
values = [f"user{n}" for n in range(82)]
|
|
result = classify_field_shape(values, field_name="username", total_events=880)
|
|
self.assertEqual(result.primary_shape, "categorical_low_cardinality")
|
|
|
|
def test_total_events_does_not_rescue_genuinely_high_cardinality_field(self):
|
|
values = [f"user{n}" for n in range(82)]
|
|
result = classify_field_shape(values, field_name="username", total_events=95)
|
|
self.assertEqual(result.primary_shape, "free_text")
|
|
|
|
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()
|