e35445da4f
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
72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fgai.history import FieldDiscoveryStore
|
|
|
|
|
|
class FieldDiscoveryConfidenceHistoryTests(unittest.TestCase):
|
|
def test_record_and_read_applied_confidence(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
store = FieldDiscoveryStore(str(Path(tmp) / "history.sqlite3"))
|
|
store.record_applied_confidence(
|
|
"custom_vendor",
|
|
["x7_field_a", "x7_field_b"],
|
|
field_confidence={"x7_field_a": 85, "x7_field_b": 40},
|
|
field_shapes={"x7_field_a": "ip", "x7_field_b": "categorical_low_cardinality"},
|
|
)
|
|
|
|
applied = store.applied_confidence("custom_vendor")
|
|
|
|
self.assertEqual(applied["x7_field_a"]["applied_confidence"], 85)
|
|
self.assertEqual(applied["x7_field_a"]["applied_shape"], "ip")
|
|
self.assertEqual(applied["x7_field_b"]["applied_confidence"], 40)
|
|
|
|
def test_record_ignores_fields_without_confidence_entry(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
store = FieldDiscoveryStore(str(Path(tmp) / "history.sqlite3"))
|
|
store.record_applied_confidence(
|
|
"custom_vendor",
|
|
["x7_field_a", "timestamp"],
|
|
field_confidence={"x7_field_a": 85},
|
|
field_shapes={"x7_field_a": "ip"},
|
|
)
|
|
|
|
applied = store.applied_confidence("custom_vendor")
|
|
|
|
self.assertIn("x7_field_a", applied)
|
|
self.assertNotIn("timestamp", applied)
|
|
|
|
def test_record_keeps_first_snapshot_and_ignores_later_calls(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
store = FieldDiscoveryStore(str(Path(tmp) / "history.sqlite3"))
|
|
store.record_applied_confidence("custom_vendor", ["x7_field_a"], {"x7_field_a": 65}, {"x7_field_a": "ip"})
|
|
store.record_applied_confidence("custom_vendor", ["x7_field_a"], {"x7_field_a": 90}, {"x7_field_a": "ip"})
|
|
|
|
applied = store.applied_confidence("custom_vendor")
|
|
|
|
self.assertEqual(applied["x7_field_a"]["applied_confidence"], 65)
|
|
self.assertEqual(len(applied), 1)
|
|
|
|
def test_confidence_history_maintenance_prunes_old_rows(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = Path(tmp) / "history.sqlite3"
|
|
store = FieldDiscoveryStore(str(db_path))
|
|
store.record_applied_confidence("custom_vendor", ["x7_field_a"], {"x7_field_a": 85}, {"x7_field_a": "ip"})
|
|
|
|
import sqlite3
|
|
old_time = int(time.time()) - 200 * 86400
|
|
with sqlite3.connect(db_path) as connection:
|
|
connection.execute("update field_confidence_history set applied_at = ?", (old_time,))
|
|
|
|
result = store.confidence_history_maintenance(retention_days=90)
|
|
|
|
self.assertEqual(result["removed"], 1)
|
|
self.assertEqual(result["remaining"], 0)
|
|
self.assertEqual(store.applied_confidence("custom_vendor"), {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|