243 lines
8.3 KiB
Python
243 lines
8.3 KiB
Python
"""Adversarial and robustness test suite for Multilingual NLP Entity Inherence Classifier.
|
|
|
|
Validates homonym disambiguation, isolated related entities, edge cases,
|
|
and CLI execution behavior via subprocess (exit codes, stream purity, JSON parsing).
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
from src.tools.models import DecisionCategory, ECPSnapshot, RelatedEntity
|
|
|
|
|
|
def test_adversarial_sao_paulo_city_vs_fc():
|
|
"""Content about city/state governance of São Paulo against ECP for São Paulo FC."""
|
|
ecp = ECPSnapshot(
|
|
target_entity_id="ent_spfc",
|
|
target_name="São Paulo Futebol Clube",
|
|
aliases=["São Paulo", "SPFC", "Tricolor Paulista"],
|
|
domain="Futebol e Esportes",
|
|
anchors=["Morumbi", "futebol", "campeonato", "Copa Libertadores", "elenco", "estádio"],
|
|
negative_anchors=[
|
|
"prefeitura de são paulo",
|
|
"governo do estado de são paulo",
|
|
"trânsito na capital paulista",
|
|
],
|
|
graph_version="1.0.0",
|
|
related_entities=[],
|
|
)
|
|
content = (
|
|
"# Obras Viárias na Capital\n\n"
|
|
"A prefeitura de São Paulo anunciou novas intervenções no trânsito na capital paulista "
|
|
"para desafogar o fluxo de veículos na região central durante os horários de pico."
|
|
)
|
|
from src.tools.classifier import InherenceClassifier
|
|
|
|
classifier = InherenceClassifier()
|
|
result = classifier.classify(ecp, content)
|
|
assert result.decision in (DecisionCategory.NOT_RELATED, DecisionCategory.TANGENTIAL)
|
|
assert result.is_inherent is False
|
|
assert result.decision != DecisionCategory.DIRECT_INHERENT
|
|
|
|
|
|
def test_adversarial_apple_fruit_recipe():
|
|
"""Content about apple fruit/culinary recipe against Apple Inc. tech entity."""
|
|
ecp = ECPSnapshot(
|
|
target_entity_id="ent_apple",
|
|
target_name="Apple",
|
|
aliases=["Apple Inc.", "Apple"],
|
|
domain="Technology",
|
|
anchors=["iPhone", "MacBook", "iOS", "silicon", "hardware"],
|
|
negative_anchors=["apple pie", "orchard harvest", "doce de maçã"],
|
|
graph_version="1.0.0",
|
|
related_entities=[],
|
|
)
|
|
content = (
|
|
"# Receita Caseira\n\n"
|
|
"Comprei maçãs frescas no mercado para preparar um doce de maçã com canela e açúcar mascavo."
|
|
)
|
|
from src.tools.classifier import InherenceClassifier
|
|
|
|
classifier = InherenceClassifier()
|
|
result = classifier.classify(ecp, content)
|
|
assert result.decision in (DecisionCategory.NOT_RELATED, DecisionCategory.TANGENTIAL)
|
|
assert result.is_inherent is False
|
|
|
|
|
|
def test_adversarial_related_entity_without_scope_context():
|
|
"""High-weight related entity mentioned in passing without required domain anchors."""
|
|
ecp = ECPSnapshot(
|
|
target_entity_id="ent_volkswagen",
|
|
target_name="Volkswagen",
|
|
aliases=["Volkswagen AG", "VW"],
|
|
domain="Automotive & Electric Vehicles",
|
|
anchors=["Elektrofahrzeuge", "Batteriezellen", "Fahrzeugproduktion"],
|
|
graph_version="1.0.0",
|
|
related_entities=[
|
|
RelatedEntity(
|
|
entity_id="ent_northvolt",
|
|
name="Northvolt",
|
|
relation_type="SUPPLIER_OF",
|
|
weight=0.95,
|
|
scope="battery_technology",
|
|
confidence=0.99,
|
|
)
|
|
],
|
|
)
|
|
# Content mentions Northvolt in an unrelated/passing architectural context without domain anchors
|
|
content = (
|
|
"# Architekturbericht aus Stockholm\n\n"
|
|
"Während unseres Stadtrundgangs besuchten wir das neue Bürogebäude von Northvolt "
|
|
"mit moderner Holzfassade und Blick auf den See."
|
|
)
|
|
from src.tools.classifier import InherenceClassifier
|
|
|
|
classifier = InherenceClassifier()
|
|
result = classifier.classify(ecp, content)
|
|
# Must be TANGENTIAL or NOT_RELATED, NEVER CONTEXTUAL_INHERENT
|
|
assert result.decision in (DecisionCategory.TANGENTIAL, DecisionCategory.NOT_RELATED)
|
|
assert result.is_inherent is False
|
|
assert result.decision != DecisionCategory.CONTEXTUAL_INHERENT
|
|
|
|
|
|
def test_adversarial_subprocess_cli_success_stdout(tmp_path):
|
|
"""Run CLI via subprocess without --output and verify stdout is pure parseable JSON."""
|
|
ecp_file = tmp_path / "ecp.json"
|
|
ecp_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"target_entity_id": "ent_petrobras",
|
|
"target_name": "Petrobras",
|
|
"aliases": ["Petrobras"],
|
|
"domain": "Oil & Gas",
|
|
"anchors": ["petróleo", "pré-sal"],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
content_file = tmp_path / "content.md"
|
|
content_file.write_text(
|
|
"# Notícia\n\nA Petrobras bateu recorde de extração de petróleo no pré-sal este mês.",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
import os
|
|
|
|
env = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONUTF8="1")
|
|
|
|
res = subprocess.run(
|
|
[sys.executable, "classify.py", "--ecp", str(ecp_file), "--content", str(content_file)],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
env=env,
|
|
)
|
|
|
|
assert res.returncode == 0
|
|
# Stdout must be directly parseable as JSON without extraneous log text
|
|
assert res.stdout is not None and len(res.stdout.strip()) > 0
|
|
parsed = json.loads(res.stdout)
|
|
assert parsed["decision"] == "DIRECT_INHERENT"
|
|
assert parsed["is_inherent"] is True
|
|
assert parsed["confidence"] >= 0.85
|
|
assert len(parsed["evidence"]) > 0
|
|
|
|
|
|
def test_adversarial_subprocess_cli_empty_content(tmp_path):
|
|
"""Run CLI via subprocess with empty content and verify error code and exit code."""
|
|
import os
|
|
|
|
env = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONUTF8="1")
|
|
|
|
ecp_file = tmp_path / "ecp.json"
|
|
ecp_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"target_entity_id": "ent_1",
|
|
"target_name": "Test",
|
|
"aliases": ["Test"],
|
|
"domain": "Tech",
|
|
"anchors": ["tech"],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
content_file = tmp_path / "empty.md"
|
|
content_file.write_text(" \n\n ", encoding="utf-8")
|
|
|
|
res = subprocess.run(
|
|
[sys.executable, "classify.py", "--ecp", str(ecp_file), "--content", str(content_file)],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
env=env,
|
|
)
|
|
|
|
assert res.returncode != 0
|
|
# Stderr must contain pure parseable error JSON
|
|
parsed_err = json.loads(res.stderr)
|
|
assert parsed_err["error_code"] == "empty_content"
|
|
|
|
|
|
def test_adversarial_subprocess_cli_missing_required_field(tmp_path):
|
|
"""Run CLI via subprocess with missing target_name and verify error payload."""
|
|
import os
|
|
|
|
env = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONUTF8="1")
|
|
|
|
ecp_file = tmp_path / "ecp_bad.json"
|
|
ecp_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"target_entity_id": "ent_1",
|
|
"aliases": ["Test"],
|
|
"domain": "Tech",
|
|
"anchors": ["tech"],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
content_file = tmp_path / "content.md"
|
|
content_file.write_text("Conteúdo de teste válido.", encoding="utf-8")
|
|
|
|
res = subprocess.run(
|
|
[sys.executable, "classify.py", "--ecp", str(ecp_file), "--content", str(content_file)],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
env=env,
|
|
)
|
|
|
|
assert res.returncode != 0
|
|
parsed_err = json.loads(res.stderr)
|
|
assert parsed_err["error_code"] == "missing_required_field"
|
|
|
|
|
|
def test_adversarial_subprocess_cli_corrupted_json(tmp_path):
|
|
"""Run CLI via subprocess with corrupted JSON and verify error payload."""
|
|
import os
|
|
|
|
env = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONUTF8="1")
|
|
|
|
ecp_file = tmp_path / "ecp_corrupted.json"
|
|
ecp_file.write_text("{ target_entity_id: not_valid_json }", encoding="utf-8")
|
|
|
|
content_file = tmp_path / "content.md"
|
|
content_file.write_text("Conteúdo de teste válido.", encoding="utf-8")
|
|
|
|
res = subprocess.run(
|
|
[sys.executable, "classify.py", "--ecp", str(ecp_file), "--content", str(content_file)],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
env=env,
|
|
)
|
|
|
|
assert res.returncode != 0
|
|
parsed_err = json.loads(res.stderr)
|
|
assert parsed_err["error_code"] == "invalid_ecp_json"
|