36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
from scripts.discover_inbound_fiscal_evidence import classify_evidence_source, discover_from_rows, safe_row_count
|
|
|
|
|
|
def test_classifies_inbound_table_from_columns_not_name():
|
|
categories = classify_evidence_source("RANDOM_TABLE", {"CODPRODUTO", "CFOP", "NCM", "CHAVE"})
|
|
|
|
assert "ITEM" in categories
|
|
assert "FISCAL_ITEM" in categories
|
|
assert "KEY" in categories
|
|
|
|
|
|
def test_discovery_output_is_sanitized_and_capability_based():
|
|
result = discover_from_rows(
|
|
[
|
|
("RANDOM_TABLE", "CODPRODUTO", 37, 40),
|
|
("RANDOM_TABLE", "XML_PAYLOAD", 261, None),
|
|
("SUPPLIER_TABLE", "CNPJ", 37, 14),
|
|
],
|
|
[("RANDOM_TABLE", "CODPRODUTO", "PRODUCTS", "CODIGO")],
|
|
)
|
|
|
|
assert result["tables"]["RANDOM_TABLE"]["xml_or_blob"] is True
|
|
assert "CNPJ" not in result["tables"]["SUPPLIER_TABLE"]["columns"]
|
|
assert result["relationships"] == [{"from_table": "RANDOM_TABLE", "from_field": "CODPRODUTO", "to_table": "PRODUCTS", "to_field": "CODIGO"}]
|
|
|
|
|
|
def test_external_table_count_is_recorded_without_aborting_database():
|
|
class Cursor:
|
|
def execute(self, sql):
|
|
raise RuntimeError("external file is not allowed")
|
|
|
|
def fetchone(self):
|
|
return None
|
|
|
|
assert safe_row_count(Cursor(), "EXTERNAL_TABLE") == {"status": "ROW_COUNT_UNAVAILABLE", "error_type": "RuntimeError"}
|