Files
wehub-resource-sync 5a558eb09e
TypeScript SDK Compatibility V1.x E2E Tests / Select Node version matrix (push) Has been cancelled
TypeScript SDK Compatibility V1.x E2E Tests / TypeScript SDK Compatibility V1.x E2E Tests Node ${{matrix.node_version}} (push) Has been cancelled
TypeScript SDK E2E Tests / TypeScript SDK E2E Tests Node ${{matrix.node_version}} (push) Has been cancelled
Opik Optimizer - E2E Tests / build-opik (push) Has been cancelled
TypeScript SDK Compatibility V1.x E2E Tests / build-opik (push) Has been cancelled
Python SDK E2E Tests / Select Python version matrix (push) Has been cancelled
Python SDK E2E Tests / Python SDK E2E Tests ${{matrix.python_version}} (push) Has been cancelled
Python SDK E2E Tests / build-opik (push) Has been cancelled
Python SDK Compatibility V1.x E2E Tests / Select Python version matrix (push) Has been cancelled
Python SDK Compatibility V1.x E2E Tests / Python SDK Compatibility V1.x E2E Tests ${{matrix.python_version}} (push) Has been cancelled
Python SDK Compatibility V1.x E2E Tests / build-opik (push) Has been cancelled
TypeScript SDK E2E Tests / Select Node version matrix (push) Has been cancelled
TypeScript SDK E2E Tests / build-opik (push) Has been cancelled
Opik Optimizer - E2E Tests / Opik Optimizer E2E Tests Python ${{matrix.python_version}} (push) Has been cancelled
Opik Optimizer - E2E Tests / Opik Optimizer Integration Smoke Tests (push) Has been cancelled
🐙 Code Quality / detect (push) Has been cancelled
🐙 Code Quality / lint (${{ matrix.leg.name }}) (push) Has been cancelled
🐙 Code Quality / summary (push) Has been cancelled
TypeScript SDK Library Integration Tests / Check Secrets (push) Has been cancelled
TypeScript SDK Library Integration Tests / opik-vercel (Vercel AI SDK / eve) (push) Has been cancelled
SDK Library Integration Tests Runner / Check Secrets (push) Has been cancelled
SDK Library Integration Tests Runner / Missed OpenAI API Key Warning (push) Has been cancelled
SDK Library Integration Tests Runner / Build (push) Has been cancelled
SDK Library Integration Tests Runner / openai_tests (push) Has been cancelled
SDK Library Integration Tests Runner / langchain_tests (push) Has been cancelled
SDK Library Integration Tests Runner / langchain_legacy_tests (push) Has been cancelled
SDK Library Integration Tests Runner / llama_index_tests (push) Has been cancelled
SDK Library Integration Tests Runner / anthropic_tests (push) Has been cancelled
SDK Library Integration Tests Runner / mistral_tests (push) Has been cancelled
SDK Library Integration Tests Runner / groq_tests (push) Has been cancelled
SDK Library Integration Tests Runner / aisuite_tests (push) Has been cancelled
SDK Library Integration Tests Runner / haystack_tests (push) Has been cancelled
SDK Library Integration Tests Runner / dspy_tests (push) Has been cancelled
SDK Library Integration Tests Runner / crewai_v0_tests (push) Has been cancelled
SDK Library Integration Tests Runner / crewai_v1_tests (push) Has been cancelled
SDK Library Integration Tests Runner / genai_tests (push) Has been cancelled
SDK Library Integration Tests Runner / adk_tests (push) Has been cancelled
SDK Library Integration Tests Runner / adk_legacy_1_3_0_tests (push) Has been cancelled
SDK Library Integration Tests Runner / evaluation_metrics_tests (push) Has been cancelled
SDK Library Integration Tests Runner / bedrock_tests (push) Has been cancelled
SDK Library Integration Tests Runner / litellm_tests (push) Has been cancelled
SDK Library Integration Tests Runner / harbor_tests (push) Has been cancelled
SDK Library Integration Tests Runner / Slack Notification (push) Has been cancelled
Lint Opik Helm Chart / render-equality (push) Has been cancelled
Opik Optimizer - Unit Tests / Opik Optimizer Unit Tests Python ${{matrix.python_version}} (push) Has been cancelled
Python BE E2E Tests / Python BE E2E (push) Has been cancelled
Python Backend Tests / run-python-backend-tests (push) Has been cancelled
Python SDK Unit Tests / Python SDK Unit Tests ${{matrix.python_version}} (push) Has been cancelled
Release Drafter / update_release_draft (push) Has been cancelled
SDK E2E Libraries Integration Tests / Check Secrets (push) Has been cancelled
SDK E2E Libraries Integration Tests / Missed OpenAI API Key Warning (push) Has been cancelled
SDK E2E Libraries Integration Tests / build-opik (push) Has been cancelled
SDK E2E Libraries Integration Tests / E2E Lib Integration Python ${{matrix.python_version}} (push) Has been cancelled
TypeScript SDK Integration Build & Publish / build-and-publish (opik-gemini) (push) Has been cancelled
TypeScript SDK Integration Build & Publish / build-and-publish (opik-langchain) (push) Has been cancelled
TypeScript SDK Integration Build & Publish / build-and-publish (opik-openai) (push) Has been cancelled
TypeScript SDK Integration Build & Publish / build-and-publish (opik-otel) (push) Has been cancelled
TypeScript SDK Integration Build & Publish / build-and-publish (opik-vercel) (push) Has been cancelled
TypeScript SDK Build & Publish / build-and-publish (push) Has been cancelled
TypeScript SDK Unit Tests / Test on Node ${{ matrix.node-version }} (push) Has been cancelled
Backend Tests / discover-tests (push) Has been cancelled
Backend Tests / ${{ matrix.name }} (push) Has been cancelled
Build and Publish SDK / build-and-publish (push) Has been cancelled
Build Opik Docker Images / set-version (push) Has been cancelled
Build Opik Docker Images / build-backend (push) Has been cancelled
Build Opik Docker Images / build-sandbox-executor-python (push) Has been cancelled
Build Opik Docker Images / build-python-backend (push) Has been cancelled
Build Opik Docker Images / build-frontend (push) Has been cancelled
Build Opik Docker Images / create-git-tag (push) Has been cancelled
ClickHouse Migration Cluster Check / validate-clickhouse-migrations (push) Has been cancelled
Docs - Publish / run (push) Has been cancelled
E2E Tests - Post Merge (v2) / 🧪 E2E v2 Tests (${{ github.event.inputs.tier || 't1' }}) (push) Has been cancelled
E2E Tests - Post Merge (v2) / 📢 Slack Notification (push) Has been cancelled
Frontend Unit Tests / Test on Node 20 (push) Has been cancelled
Guardrails E2E Tests / Select Python version matrix (push) Has been cancelled
Guardrails E2E Tests / Guardrails E2E Tests ${{matrix.python_version}} (push) Has been cancelled
Guardrails E2E Tests / 📢 Slack Notification (push) Has been cancelled
Guardrails Backend Unit Tests / Guardrails Backend Unit Tests (push) Has been cancelled
Guardrails Backend Unit Tests / 📢 Slack Notification (push) Has been cancelled
Lint Opik Helm Chart / lint-helm-chart (Helm v3.21.0) (push) Has been cancelled
Lint Opik Helm Chart / lint-helm-chart (Helm v4.2.0) (push) Has been cancelled
Lint Opik Helm Chart / unittest-helm-chart (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:25:44 +08:00

319 lines
11 KiB
Python

import pytest
from opik.anonymizer import rules
from opik.anonymizer import rules_anonymizer
class TestRulesAnonymizer:
"""Test suite for RulesAnonymizer functionality."""
def test_init(self):
"""Test RulesAnonymizer initialization."""
regex_rule = rules.RegexRule(r"\d+", "***")
rules_ = [regex_rule]
anonymizer = rules_anonymizer.RulesAnonymizer(rules_, max_depth=5)
assert anonymizer.rules == rules_
assert anonymizer.max_depth == 5
def test_anonymize_text__single_regex_rule(self):
"""Test anonymizing text with a single regex rule."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
result = anonymizer.anonymize_text("Phone: 123-456-7890")
assert result == "Phone: ***-***-***"
def test_anonymize_text__multiple_regex_rules(self):
"""Test anonymizing text with multiple regex rules applied sequentially."""
email_rule = rules.RegexRule(
r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "[EMAIL]"
)
phone_rule = rules.RegexRule(r"\d{3}-\d{3}-\d{4}", "[PHONE]")
anonymizer = rules_anonymizer.RulesAnonymizer([email_rule, phone_rule])
result = anonymizer.anonymize_text("Contact: user@example.com or 123-456-7890")
assert result == "Contact: [EMAIL] or [PHONE]"
def test_anonymize_text__function_rule(self):
"""Test anonymizing text with a function rule."""
def uppercase_anonymizer(text: str) -> str:
return text.upper()
function_rule = rules.FunctionRule(uppercase_anonymizer)
anonymizer = rules_anonymizer.RulesAnonymizer([function_rule])
result = anonymizer.anonymize_text("hello world")
assert result == "HELLO WORLD"
def test_anonymize_text__mixed_rules(self):
"""Test anonymizing text with mixed rule types."""
regex_rule = rules.RegexRule(r"\d+", "XXX")
function_rule = rules.FunctionRule(lambda text: text.replace(" ", "_"))
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule, function_rule])
result = anonymizer.anonymize_text("Phone 123 456")
assert result == "Phone_XXX_XXX"
def test_anonymize_text__no_rules(self):
"""Test anonymizing text with no rules returns original text."""
anonymizer = rules_anonymizer.RulesAnonymizer([])
result = anonymizer.anonymize_text("sensitive data")
assert result == "sensitive data"
def test_anonymize__string_data(self):
"""Test anonymizing string data."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
result = anonymizer.anonymize("Phone: 123-456-7890")
assert result == "Phone: ***-***-***"
def test_anonymize__dict_data(self):
"""Test anonymizing dictionary data."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
data = {"name": "John Doe", "phone": "123-456-7890", "address": "123 Main St"}
result = anonymizer.anonymize(data)
expected = {
"name": "John Doe",
"phone": "***-***-***",
"address": "*** Main St",
}
assert result == expected
def test_anonymize__list_data(self):
"""Test anonymizing list data."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
data = ["Contact: 123-456-7890", "Another: 987-654-3210", "No numbers here"]
result = anonymizer.anonymize(data)
expected = ["Contact: ***-***-***", "Another: ***-***-***", "No numbers here"]
assert result == expected
def test_anonymize__nested_dict_data(self):
"""Test anonymizing nested dictionary data."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
data = {
"user": {
"name": "John Doe",
"contact": {"phone": "123-456-7890", "email": "john@example.com"},
},
"id": "12345",
}
result = anonymizer.anonymize(data)
expected = {
"user": {
"name": "John Doe",
"contact": {"phone": "***-***-***", "email": "john@example.com"},
},
"id": "***",
}
assert result == expected
def test_anonymize__nested_list_data(self):
"""Test anonymizing nested list data."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
data = [
["Phone: 123", "Email: user@test.com"],
{"contact": "456-789-0123"},
"Simple string with 999",
]
result = anonymizer.anonymize(data)
expected = [
["Phone: ***", "Email: user@test.com"],
{"contact": "***-***-***"},
"Simple string with ***",
]
assert result == expected
def test_anonymize__mixed_nested_data(self):
"""Test anonymizing complex mixed nested data structures."""
email_rule = rules.RegexRule(
r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "[EMAIL]"
)
phone_rule = rules.RegexRule(r"\d{3}-\d{3}-\d{4}", "[PHONE]")
anonymizer = rules_anonymizer.RulesAnonymizer([email_rule, phone_rule])
data = {
"users": [
{"name": "John Doe", "contacts": ["john@example.com", "123-456-7890"]},
{"name": "Jane Smith", "contacts": ["jane@test.org", "987-654-3210"]},
],
"admin_contact": "admin@company.com",
}
result = anonymizer.anonymize(data)
expected = {
"users": [
{"name": "John Doe", "contacts": ["[EMAIL]", "[PHONE]"]},
{"name": "Jane Smith", "contacts": ["[EMAIL]", "[PHONE]"]},
],
"admin_contact": "[EMAIL]",
}
assert result == expected
def test_anonymize__non_string_values_unchanged(self):
"""Test that non-string values are not processed."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
data = {
"number": 12345,
"float_val": 123.45,
"bool_val": True,
"none_val": None,
"string_val": "123",
}
result = anonymizer.anonymize(data)
expected = {
"number": 12345,
"float_val": 123.45,
"bool_val": True,
"none_val": None,
"string_val": "***",
}
assert result == expected
def test_anonymize__max_depth_limiting(self):
"""Test that max_depth limits recursion depth."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule], max_depth=2)
# Create deeply nested data that exceeds max_depth
data = {"level1": {"level2": {"level3": {"phone": "123-456-7890"}}}}
result = anonymizer.anonymize(data)
# At max_depth=2, level3 should not be processed
expected = {
"level1": {
"level2": {
"level3": {
"phone": "123-456-7890" # Should remain unchanged
}
}
}
}
assert result == expected
def test_anonymize__max_depth_exact_limit(self):
"""Test that max_depth allows processing exactly at the limit."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule], max_depth=4)
# Create data where string is processed at depth=3 (within limit)
data = {"level1": {"level2": {"phone": "123-456-7890"}}}
result = anonymizer.anonymize(data)
# At max_depth=4, the string at depth 3 should be processed
expected = {"level1": {"level2": {"phone": "***-***-***"}}}
assert result == expected
def test_anonymize__circular_reference_protection(self):
"""Test protection against circular references through depth limiting."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule], max_depth=3)
# Create circular reference
data = {"phone": "123"}
data["self"] = data
# Should not cause infinite recursion due to max_depth
result = anonymizer.anonymize(data)
# The result should have processed the string but stopped at max depth
assert result["phone"] == "***"
assert "self" in result
@pytest.mark.parametrize("rule_count", [1, 5, 10])
def test_anonymize_text__performance_with_multiple_rules(self, rule_count):
"""Test performance with varying numbers of rules."""
rules_ = []
for i in range(rule_count):
rules_.append(rules.RegexRule(f"pattern{i}", f"replacement{i}"))
anonymizer = rules_anonymizer.RulesAnonymizer(rules_)
# Test with a simple string
result = anonymizer.anonymize_text("test pattern0 and pattern5")
# Should apply all applicable rules
assert "replacement0" in result or "pattern0" in result
def test_anonymize_text__empty_string_handling(self):
"""Test handling of empty strings."""
regex_rule = rules.RegexRule(r"\d+", "***")
anonymizer = rules_anonymizer.RulesAnonymizer([regex_rule])
result = anonymizer.anonymize_text("")
assert result == ""
def test_anonymize_text__complex_regex_patterns(self):
"""Test complex regex patterns for real-world scenarios."""
# Email pattern
email_rule = rules.RegexRule(
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL_REDACTED]"
)
# Credit card pattern (simplified)
cc_rule = rules.RegexRule(
r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b", "[CARD_REDACTED]"
)
# SSN pattern
ssn_rule = rules.RegexRule(r"\b\d{3}-\d{2}-\d{4}\b", "[SSN_REDACTED]")
anonymizer = rules_anonymizer.RulesAnonymizer([email_rule, cc_rule, ssn_rule])
text = """
Contact info:
Email: john.doe@example.com
Credit Card: 1234 5678 9012 3456
SSN: 123-45-6789
"""
result = anonymizer.anonymize_text(text)
assert result != text
assert (
result
== """
Contact info:
Email: [EMAIL_REDACTED]
Credit Card: [CARD_REDACTED]
SSN: [SSN_REDACTED]
"""
)