chore: establish project baseline

This commit is contained in:
OpenCode
2026-09-21 15:51:47 -03:00
commit 9dcc263217
459 changed files with 447890 additions and 0 deletions
+183
View File
@@ -0,0 +1,183 @@
"""Mission 00D Phase 8: prove arbitrary SQL cannot reach execution through
the closed catalog API, and that legitimate Firebird constructs used by
the project's own internal queries are not rejected.
``FirebirdCatalog`` has no ``execute``/``query``/``raw_execute`` method at
all, so most classic SQL-injection payloads simply have nowhere to go --
these tests prove that structurally (no method exists to try them on) and
prove that the one dynamic operation, ``sample_relation_rows``, rejects
every malicious relation/column value because it is validated against the
catalog itself, never interpreted as SQL text.
"""
import pytest
from gti_discovery.catalog import MAX_SAMPLE_ROWS, FirebirdCatalog, _quote_identifier
from gti_discovery.firebird_queries import _ADAPTER_REGISTRATION_TOKEN, _register_approved_adapter
from gti_discovery.read_only import validate_read_only
class _RecordingAdapter:
read_only = True
def __init__(self, rows_by_call=None):
self.calls: list[tuple[str, tuple]] = []
self._rows_by_call = rows_by_call or {}
def execute(self, sql, params=()):
self.calls.append((sql, params))
return self._rows_by_call.get(len(self.calls) - 1, [])
_register_approved_adapter(_RecordingAdapter, _ADAPTER_REGISTRATION_TOKEN)
def _catalog_with_relations(relations, fields_by_relation):
"""A FirebirdCatalog whose relations/fields queries return fixture rows."""
relation_rows = [{"RELATION_NAME": name, "RELATION_ID": i, "SYSTEM_FLAG": 0} for i, name in enumerate(relations)]
field_rows = [
{"RELATION_NAME": relation, "FIELD_NAME": field, "FIELD_POSITION": i}
for relation, fields in fields_by_relation.items()
for i, field in enumerate(fields)
]
adapter = _RecordingAdapter({0: relation_rows, 1: field_rows})
catalog = FirebirdCatalog(adapter)
return catalog, adapter
# --- Phase 8: closed API surface -----------------------------------------
def test_no_arbitrary_sql_method_exists_at_all():
for forbidden in ("execute", "query", "raw_execute", "execute_statement", "run_sql", "run_query"):
assert not hasattr(FirebirdCatalog, forbidden)
# --- Phase 8: relation/column must be catalog-known, never free text -----
@pytest.mark.parametrize("malicious_relation", [
"PRODUCTS\"; DROP TABLE X; --",
"PRODUCTS' OR '1'='1",
'PRODUCTS" ("EAN"',
"PRODUCTS/*comment*/",
"PRODUCTS UNION SELECT * FROM RDB$USERS",
"SOME_PROCEDURE(1,2,3)",
'"SOME_UDF"(1)',
"PRODUCTS; SELECT * FROM RDB$USERS",
"(SELECT PASSWORD FROM USERS)",
"PRODUCTS AS X UNION SELECT 1",
])
def test_sample_relation_rows_rejects_unknown_or_malicious_relation(malicious_relation):
catalog, adapter = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID", "EAN"]})
with pytest.raises(ValueError):
catalog.sample_relation_rows(malicious_relation, ["ID"], 1)
# never reached the driver with a sampling query (only the relations
# bootstrap call, if any, is allowed to have happened)
sampling_calls = [c for c in adapter.calls if "FIRST" in c[0].upper()]
assert sampling_calls == []
@pytest.mark.parametrize("malicious_column", [
"EAN\"; DROP TABLE X; --",
"EAN' OR '1'='1",
"EAN, PASSWORD",
"EAN) UNION SELECT PASSWORD FROM USERS --",
"COUNT(*)",
"(SELECT 1)",
])
def test_sample_relation_rows_rejects_unknown_or_malicious_column(malicious_column):
catalog, adapter = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID", "EAN"]})
with pytest.raises(ValueError):
catalog.sample_relation_rows("PRODUCTS", [malicious_column], 1)
sampling_calls = [c for c in adapter.calls if "FIRST" in c[0].upper()]
assert sampling_calls == []
def test_sample_relation_rows_accepts_legitimate_relation_and_columns():
catalog, adapter = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID", "EAN"]})
catalog.sample_relation_rows("PRODUCTS", ["ID", "EAN"], 5)
sql, params = adapter.calls[-1]
assert sql == 'SELECT FIRST 5 "ID", "EAN" FROM "PRODUCTS"'
assert params == ()
@pytest.mark.parametrize("limit", [0, -1, 51, 10 ** 9, "5", 5.0, True])
def test_sample_relation_rows_rejects_bad_limit(limit):
catalog, _ = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID"]})
with pytest.raises(ValueError):
catalog.sample_relation_rows("PRODUCTS", ["ID"], limit)
def test_sample_relation_rows_limit_ceiling_is_enforced():
catalog, _ = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID"]})
catalog.sample_relation_rows("PRODUCTS", ["ID"], MAX_SAMPLE_ROWS)
with pytest.raises(ValueError):
catalog.sample_relation_rows("PRODUCTS", ["ID"], MAX_SAMPLE_ROWS + 1)
def test_sample_relation_rows_never_accepts_empty_columns():
catalog, _ = _catalog_with_relations(["PRODUCTS"], {"PRODUCTS": ["ID"]})
with pytest.raises(ValueError):
catalog.sample_relation_rows("PRODUCTS", [], 1)
def test_quote_identifier_doubles_embedded_quotes():
assert _quote_identifier('WEIRD"NAME') == '"WEIRD""NAME"'
# --- Phase 5: procedures are metadata-only, never executable -------------
def test_get_procedures_returns_metadata_never_a_callable():
adapter = _RecordingAdapter({0: [{"PROCEDURE_NAME": "SP_RECALC_STOCK", "PROCEDURE_ID": 1}]})
catalog = FirebirdCatalog(adapter)
procedures = catalog.get_procedures()
assert procedures == [{"PROCEDURE_NAME": "SP_RECALC_STOCK", "PROCEDURE_ID": 1}]
# a procedure name is never itself a relation, so it can never be
# threaded into sample_relation_rows
with pytest.raises(ValueError):
catalog.sample_relation_rows("SP_RECALC_STOCK", ["ID"], 1)
# --- read_only.py kept as an offline tool: bypass tests still apply ------
@pytest.mark.parametrize("payload", [
'"SOME_PROCEDURE"(1,2,3)',
'SELECT * FROM "SOME_PROC"(1)',
"SELECT * FROM PRODUCTS; DROP TABLE PRODUCTS",
"SELECT * FROM PRODUCTS -- comment",
"SELECT * FROM PRODUCTS /* comment */",
"EXECUTE PROCEDURE SP_DANGEROUS",
"SELECT SOME_UDF(1) FROM RDB$DATABASE",
"SELECT (SELECT PASSWORD FROM USERS) FROM PRODUCTS",
])
def test_offline_validator_still_rejects_every_bypass_payload(payload):
with pytest.raises(ValueError):
validate_read_only(payload)
@pytest.mark.parametrize("legit", [
"SELECT CAST(X AS INTEGER) FROM PRODUCTS",
"SELECT COUNT(*) FROM PRODUCTS",
"SELECT SUM(QTY), MIN(QTY), MAX(QTY), AVG(QTY) FROM STOCK GROUP BY PRODUCT_ID",
"SELECT COALESCE(X, 0) FROM PRODUCTS",
"WITH x AS (SELECT 1 AS n FROM RDB$DATABASE) SELECT n FROM x",
# UNION of two plain read-only SELECTs has no side effect -- the
# validator correctly accepts it rather than over-blocking.
"SELECT ID FROM A UNION SELECT ID FROM B",
# A quoted alias containing reserved-word-looking text or a semicolon
# is inert -- it can never execute anything, it is just a name; the
# masking correctly treats quoted content as opaque, not as SQL.
"SELECT * FROM PRODUCTS AS \"'; DROP TABLE X; --\"",
# A string literal containing reserved words is data, not SQL --
# masking must not accidentally treat it as a statement.
"SELECT 'reserved SELECT INSERT DELETE word inside a string' FROM PRODUCTS",
])
def test_offline_validator_accepts_legitimate_firebird_constructs(legit):
assert validate_read_only(legit) == legit
def test_offline_validator_is_not_wired_into_the_live_catalog_path():
"""Mission 00D: read_only.validate_read_only must not be imported by
catalog.py -- it is an optional offline tool, never the runtime gate."""
import gti_discovery.catalog as catalog_module
assert "validate_read_only" not in dir(catalog_module)
assert "read_only" not in catalog_module.__dict__
+291
View File
@@ -0,0 +1,291 @@
import pytest
from gti_discovery.catalog import FirebirdCatalog
from gti_discovery.firebird_queries import _ADAPTER_REGISTRATION_TOKEN, _register_approved_adapter
from gti_discovery.collector import ReadOnlyCatalogReader, _CATALOG_QUERIES, _FIXTURE_REGISTRATION_TOKEN, _approved_fixture_reader, _fetch_rows, _is_optional_unsupported, _safe_value, collect_metadata
from gti_discovery.fingerprint import build_fingerprint
from gti_discovery.read_only import validate_read_only
@_approved_fixture_reader(token=_FIXTURE_REGISTRATION_TOKEN)
class CatalogFixture(ReadOnlyCatalogReader):
def __init__(self, failures=None, missing=None):
self.failures = failures or {}
self.missing = missing or set()
self.calls = []
def _rows(self, name):
self.calls.append(name)
if name in self.failures:
raise self.failures[name]
return [{"NAME": name, "POSITION": 0}]
def __getattribute__(self, name):
if name.startswith("list_") and name.removeprefix("list_") in object.__getattribute__(self, "missing"):
raise AttributeError(name)
return object.__getattribute__(self, name)
def list_database(self): return self._rows("database")
def list_relations(self): return self._rows("relations")
def list_relation_fields(self): return self._rows("fields")
def list_keys(self): return self._rows("keys")
def list_indexes(self): return self._rows("indexes")
def list_foreign_keys(self): return self._rows("foreign_keys")
def list_triggers(self): return self._rows("triggers")
def list_procedures(self): return self._rows("procedures")
def list_procedure_parameters(self): return self._rows("procedure_parameters")
def list_packages(self): return self._rows("packages")
def list_generators(self): return self._rows("generators")
def list_dependencies(self): return self._rows("dependencies")
def list_monitoring(self): return self._rows("monitoring")
def test_optional_catalogs_are_collected_independently():
metadata = collect_metadata(CatalogFixture())
assert metadata["generators"] == [{"NAME": "generators", "POSITION": 0}]
assert metadata["dependencies"] == [{"NAME": "dependencies", "POSITION": 0}]
assert metadata["monitoring"] == [{"NAME": "monitoring", "POSITION": 0}]
def test_optional_unsupported_catalog_is_unavailable_but_permission_is_an_error():
reader = CatalogFixture(
{
"packages": RuntimeError("Table unknown RDB$PACKAGES"),
"generators": RuntimeError("no permission for SELECT"),
}
)
metadata = collect_metadata(reader)
assert "packages" in metadata["unavailable"]
assert {error["kind"] for error in metadata["errors"]} == {"optional_unsupported", "query_error"}
assert {error["query"] for error in metadata["errors"]} == {"generators", "packages"}
@pytest.mark.parametrize("message", [
"permission denied: catalog unsupported",
"insufficient privilege; feature not supported",
"access denied: table unknown",
])
def test_permission_failures_never_count_as_optional_unsupported(message):
assert not _is_optional_unsupported("packages", RuntimeError(message))
@pytest.mark.parametrize("error", [PermissionError(), PermissionError("catalog unsupported")])
def test_permission_error_type_is_never_optional_unsupported(error):
assert not _is_optional_unsupported("packages", error)
def test_empty_permission_error_is_recorded_as_query_error():
metadata = collect_metadata(CatalogFixture({"packages": PermissionError()}))
assert metadata["errors"] == [{"query": "packages", "kind": "query_error", "error": "PermissionError"}]
def test_missing_optional_reader_method_does_not_raise_during_collection():
metadata = collect_metadata(CatalogFixture(missing={"packages"}))
assert "packages" in metadata["unavailable"]
def test_firebird_catalog_implements_closed_interface_explicitly():
class Adapter:
read_only = True
def execute(self, sql, params):
return []
_register_approved_adapter(Adapter, _ADAPTER_REGISTRATION_TOKEN)
reader = FirebirdCatalog(Adapter())
assert isinstance(reader, ReadOnlyCatalogReader)
assert reader.list_packages() == []
def test_firebird_catalog_exposes_no_arbitrary_sql_method():
"""Mission 00D: the closed API must never offer execute/query/raw_execute."""
class Adapter:
read_only = True
def execute(self, sql, params):
return []
_register_approved_adapter(Adapter, _ADAPTER_REGISTRATION_TOKEN)
reader = FirebirdCatalog(Adapter())
for forbidden in ("execute", "query", "raw_execute", "execute_statement", "run_sql"):
assert not hasattr(reader, forbidden), f"FirebirdCatalog must not expose {forbidden!r}"
def test_firebird_catalog_rejects_adapter_without_proven_read_only():
class Unproven:
read_only = False
def execute(self, sql, params):
return []
_register_approved_adapter(Unproven, _ADAPTER_REGISTRATION_TOKEN)
with pytest.raises(ValueError):
FirebirdCatalog(Unproven())
def test_firebird_catalog_hides_raw_handle_from_attribute_access():
class Adapter:
read_only = True
def execute(self, sql, params):
return []
_register_approved_adapter(Adapter, _ADAPTER_REGISTRATION_TOKEN)
reader = FirebirdCatalog(Adapter())
with pytest.raises(AttributeError):
reader._handle
@pytest.mark.parametrize("failure", [NotImplementedError(), RuntimeError("catalog unsupported")])
def test_explicit_optional_capability_failures_are_unavailable(failure):
metadata = collect_metadata(CatalogFixture({"monitoring": failure}))
assert "monitoring" in metadata["unavailable"]
assert metadata["errors"][0]["kind"] == "optional_unsupported"
def test_tuple_row_values_are_sanitized_field_by_field():
class UnknownValue:
pass
row = ("PRODUCT", 0, b"binary", "password=hidden", "x" * 2048, UnknownValue())
assert _safe_value(row) == ["PRODUCT", 0, "<redacted>", "<redacted>", "<redacted>", "<redacted>"]
def test_db_api_tuple_rows_use_cursor_description_for_metadata_names():
class Cursor:
description = [("RELATION_NAME",), ("FIELD_POSITION",)]
def fetchall(self):
return [("PRODUCT", 0)]
assert _fetch_rows(Cursor()) == [{"RELATION_NAME": "PRODUCT", "FIELD_POSITION": 0}]
def test_all_catalog_queries_are_select_only_and_source_free():
for query in _CATALOG_QUERIES.values():
validate_read_only(query)
assert not any(token in query.upper() for token in ("RDB$TRIGGER_SOURCE", "RDB$PROCEDURE_SOURCE", "RDB$PACKAGE_HEADER_SOURCE", "RDB$PACKAGE_BODY_SOURCE", "RDB$DEFAULT_SOURCE"))
def test_missing_optional_reader_is_recorded_as_unavailable():
reader = CatalogFixture(missing={"packages"})
metadata = collect_metadata(reader)
assert "packages" in metadata["unavailable"]
def test_collect_metadata_rejects_duck_typed_non_reader():
"""Mission 00D Phase 9: collect_metadata must not accept an
arbitrary object just because it happens to expose the same method
names -- ReadOnlyCatalogReader is an ABC and this is a runtime
check (isinstance), not merely a type hint a caller could ignore."""
class DuckTyped:
def get_database_info(self): return []
def list_relations(self): return []
def list_relation_fields(self): return []
def list_keys(self): return []
def list_indexes(self): return []
def list_foreign_keys(self): return []
def list_triggers(self): return []
def list_procedures(self): return []
def list_procedure_parameters(self): return []
def list_packages(self): return []
with pytest.raises(TypeError, match="reader must be ReadOnlyCatalogReader"):
collect_metadata(DuckTyped())
def test_collect_metadata_accepts_real_firebird_catalog():
class Adapter:
read_only = True
def execute(self, sql, params=()):
return []
_register_approved_adapter(Adapter, _ADAPTER_REGISTRATION_TOKEN)
metadata = collect_metadata(FirebirdCatalog(Adapter()))
assert metadata["database"] == []
def test_collection_row_order_is_deterministic_for_reversed_reader_rows():
@_approved_fixture_reader(token=_FIXTURE_REGISTRATION_TOKEN)
class ReorderedReader(ReadOnlyCatalogReader):
def __init__(self, reverse):
self.reverse = reverse
def _rows(self, name):
rows = {
"database": [{"MON$DATABASE_NAME": "DB", "MON$ODS_MAJOR": 13}],
"relations": [{"RELATION_NAME": "B", "RELATION_ID": 2}, {"RELATION_NAME": "A", "RELATION_ID": 1}],
"fields": [
{"RELATION_NAME": "B", "FIELD_NAME": "Z", "FIELD_POSITION": 1, "TYPE_NAME": "INTEGER"},
{"RELATION_NAME": "A", "FIELD_NAME": "ID", "FIELD_POSITION": 0, "TYPE_NAME": "INTEGER"},
],
"keys": [{"RELATION_NAME": "B", "CONSTRAINT_TYPE": "PRIMARY KEY", "FIELD_NAME": "ID"}],
"indexes": [{"RELATION_NAME": "A", "INDEX_NAME": "IX_A", "FIELD_NAME": "ID"}],
"foreign_keys": [{"CHILD_RELATION_NAME": "B", "REFERENCED_RELATION_NAME": "A", "CHILD_FIELD_NAME": "A_ID"}],
"triggers": [{"TRIGGER_NAME": "TR_B", "RELATION_NAME": "B", "TRIGGER_TYPE": 1}],
"procedures": [{"PROCEDURE_NAME": "P_B", "PROCEDURE_ID": 2}, {"PROCEDURE_NAME": "P_A", "PROCEDURE_ID": 1}],
"procedure_parameters": [{"PROCEDURE_NAME": "P_B", "PARAMETER_NUMBER": 1}],
"packages": [{"PACKAGE_NAME": "PKG_B", "PACKAGE_ID": 2}],
"generators": [{"GENERATOR_NAME": "GEN_B", "GENERATOR_ID": 2}],
"dependencies": [{"DEPENDENT_NAME": "P_B", "DEPENDED_ON_NAME": "B"}],
"monitoring": [(2, "running"), (1, "idle")],
}[name]
return list(reversed(rows)) if self.reverse else rows
def list_database(self): return self._rows("database")
def list_relations(self): return self._rows("relations")
def list_relation_fields(self): return self._rows("fields")
def list_keys(self): return self._rows("keys")
def list_indexes(self): return self._rows("indexes")
def list_foreign_keys(self): return self._rows("foreign_keys")
def list_triggers(self): return self._rows("triggers")
def list_procedures(self): return self._rows("procedures")
def list_procedure_parameters(self): return self._rows("procedure_parameters")
def list_packages(self): return self._rows("packages")
def list_generators(self): return self._rows("generators")
def list_dependencies(self): return self._rows("dependencies")
def list_monitoring(self): return self._rows("monitoring")
assert collect_metadata(ReorderedReader(False)) == collect_metadata(ReorderedReader(True))
def test_collect_metadata_rejects_structured_operation_overrides():
class OverriddenReader(ReadOnlyCatalogReader):
def get_database_info(self):
return []
with pytest.raises(TypeError, match="sealed structured operations"):
collect_metadata(OverriddenReader())
def test_approved_fixture_registration_does_not_authorize_subclasses():
class DerivedFixture(CatalogFixture):
pass
with pytest.raises(TypeError, match="sealed structured operations"):
collect_metadata(DerivedFixture())
def test_catalog_query_map_is_not_exported_from_collector():
import gti_discovery.collector as collector
assert not hasattr(collector, "METADATA_QUERIES")
def test_engine_context_changes_fingerprint():
base = {"database": [{"ODS_MAJOR": 12}], "engine": {"dialect": 3}}
changed = {"database": [{"ODS_MAJOR": 12}], "engine": {"dialect": 1}}
assert build_fingerprint(base) != build_fingerprint(changed)
+28
View File
@@ -0,0 +1,28 @@
from gti_discovery.company_context import collect_company_context
class Reader:
def company_context_rows(self):
return [
{"UF": "SP", "REGIME": "SIMPLES NACIONAL", "CNPJ": "12.345.678/0001-90"},
{"UF": "SP", "REGIME": "SIMPLES NACIONAL", "CNPJ": "12.345.678/0001-90"},
]
def test_company_context_is_targeted_and_redacts_cnpj():
result = collect_company_context(Reader())
assert result == {"uf": "SP", "regime": "SIMPLES NACIONAL", "cnpj": "REDACTED", "record_count": 2}
def test_company_context_rejects_generic_readers():
class Generic:
def execute(self, sql):
return []
try:
collect_company_context(Generic())
except TypeError as error:
assert "company_context_rows" in str(error)
else:
raise AssertionError("generic query access must be rejected")
+141
View File
@@ -0,0 +1,141 @@
import importlib.util
from pathlib import Path
SCRIPT = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
def load_script():
spec = importlib.util.spec_from_file_location("run_firebird_discovery_dependency", SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def valid_fdb():
return type(
"Fdb",
(),
{
"__version__": "2.0.4",
"ISOLATION_LEVEL_READ_COMMITED_RO": b"\x03\x08\x06\x0f\x11",
},
)()
def patch_valid_environment(monkeypatch, module, tmp_path):
client = tmp_path / "fbclient.dll"
client.write_bytes(b"client")
monkeypatch.setattr(module, "CLIENT_LIBRARY", client)
monkeypatch.setattr(module, "validate_fdb_provenance", lambda: {
"status": "VALID",
"distribution_name": "fdb",
"distribution_version": "2.0.4",
})
monkeypatch.setattr(module.importlib.metadata, "version", lambda name: "2.0.4")
monkeypatch.setattr(module, "_client_metadata", lambda path: {
"path": str(path),
"exists": True,
"version": "2.5.9.27139",
"architecture": "x64",
})
return client
def test_validate_firebird_runtime_returns_complete_compatibility_evidence(monkeypatch, tmp_path):
module = load_script()
patch_valid_environment(monkeypatch, module, tmp_path)
evidence = module.validate_firebird_runtime(fdb_module=valid_fdb())
assert evidence["status"] == "VALID"
assert evidence["python_runtime"]["architecture"] == "x64"
assert evidence["fdb"]["version"] == "2.0.4"
assert evidence["client"]["version"] == "2.5.9.27139"
assert evidence["client"]["architecture"] == "x64"
assert evidence["compatibility"] == {
"status": "VALID",
"python_client_architecture_match": True,
"fdb_client_version_match": True,
"read_only_api": True,
}
def test_validate_firebird_runtime_without_module_is_unknown(monkeypatch, tmp_path):
module = load_script()
patch_valid_environment(monkeypatch, module, tmp_path)
evidence = module.validate_firebird_runtime()
assert evidence["status"] == "UNKNOWN"
assert evidence["compatibility"]["read_only_api"] is False
assert evidence["blocker"] == "compatibility_unproven"
def test_preimport_gate_requires_client_architecture(monkeypatch, tmp_path):
module = load_script()
client = tmp_path / "fbclient.dll"
client.write_bytes(b"client")
monkeypatch.setattr(module, "CLIENT_LIBRARY", client)
monkeypatch.setattr(module, "validate_fdb_provenance", lambda: {"status": "VALID"})
monkeypatch.setattr(module, "_client_metadata", lambda path: {
"path": str(path), "exists": True, "version": "2.5.9.27139"
})
monkeypatch.setattr(module, "_read_client_architecture", lambda path: "UNKNOWN")
monkeypatch.setattr(module.importlib.metadata, "version", lambda name: "2.0.4")
try:
module._validate_preimport_compatibility()
except RuntimeError as error:
assert "compatibility" in str(error)
else:
raise AssertionError("unknown client architecture must block before import")
def test_preimport_gate_requires_exact_read_only_api(monkeypatch, tmp_path):
module = load_script()
patch_valid_environment(monkeypatch, module, tmp_path)
monkeypatch.setattr(module, "_preimport_read_only_api_proven", lambda: False)
try:
module._validate_preimport_compatibility()
except RuntimeError as error:
assert "compatibility" in str(error)
else:
raise AssertionError("unproven read-only API must block before import")
def test_unknown_client_architecture_blocks(monkeypatch, tmp_path):
module = load_script()
patch_valid_environment(monkeypatch, module, tmp_path)
monkeypatch.setattr(module, "_client_metadata", lambda path: {
"path": str(path), "exists": True, "version": "2.5.9.27139", "architecture": "UNKNOWN"
})
evidence = module.validate_firebird_runtime(fdb_module=valid_fdb())
assert evidence["status"] == "UNKNOWN"
assert evidence["blocker"] == "compatibility_unproven"
def test_unknown_client_version_blocks(monkeypatch, tmp_path):
module = load_script()
patch_valid_environment(monkeypatch, module, tmp_path)
monkeypatch.setattr(module, "_client_metadata", lambda path: {
"path": str(path), "exists": True, "version": None, "architecture": "x64"
})
evidence = module.validate_firebird_runtime(fdb_module=valid_fdb())
assert evidence["status"] == "UNKNOWN"
assert evidence["blocker"] == "compatibility_unproven"
def test_unsupported_read_only_api_is_unknown(monkeypatch, tmp_path):
module = load_script()
patch_valid_environment(monkeypatch, module, tmp_path)
fdb = type("Fdb", (), {"__version__": "2.0.4"})()
evidence = module.validate_firebird_runtime(fdb_module=fdb)
assert evidence["status"] == "UNKNOWN"
assert evidence["blocker"] == "compatibility_unproven"
+26
View File
@@ -0,0 +1,26 @@
import json
from pathlib import Path
import pytest
DOCS = Path(__file__).resolve().parents[1] / "docs"
FMAP = DOCS / "ERP_FISCAL_FIELD_MAP_V1.json"
pytestmark = pytest.mark.skipif(not FMAP.exists(), reason="field map artifact not generated")
WRITABLE = {"IPI", "EX_TIPI", "CEST"}
def test_only_proven_product_fields_are_write_candidates():
fields = json.loads(FMAP.read_text(encoding="utf-8"))["fields"]
for f in fields:
if f["write_candidate"]:
assert f["table"] == "C000025" and f["field"] in WRITABLE and f["confidence"] == "HIGH"
if f["usage_class"] in ("LEGACY_UNUSED", "UNKNOWN") or f["confidence"] == "LOW":
assert not f["write_candidate"]
def test_every_context_input_has_a_known_status():
doc = json.loads((DOCS / "FISCAL_CONTEXT_REQUIREMENTS_V1.json").read_text(encoding="utf-8"))
allowed = {"AVAILABLE_FROM_ERP", "AVAILABLE_FROM_AGENT_CONFIG", "AVAILABLE_FROM_JOB", "MISSING"}
for dom in doc["domains"].values():
assert {i["status"] for i in dom["inputs"].values()} <= allowed
+274
View File
@@ -0,0 +1,274 @@
import json
import pytest
from gti_discovery.collector import (
_CATALOG_QUERIES,
_FIXTURE_REGISTRATION_TOKEN,
_approved_fixture_reader,
ReadOnlyCatalogReader,
_safe_value,
collect_metadata,
)
from gti_discovery.read_only import validate_read_only
from gti_discovery.fingerprint import (
SCHEMA_CHANGED,
SCHEMA_SUPPORTED,
SCHEMA_UNKNOWN,
build_fingerprint,
classify_schema,
)
def test_fingerprint_is_order_independent():
a = {"tables": [{"name": "A", "columns": ["ID", "NAME"]}]}
b = {"tables": [{"columns": ["ID", "NAME"], "name": "A"}]}
assert build_fingerprint(a) == build_fingerprint(b)
def test_relevant_schema_change_changes_fingerprint():
a = {"tables": [{"name": "A", "columns": ["ID"]}]}
b = {"tables": [{"name": "A", "columns": ["ID", "NCM"]}]}
assert build_fingerprint(a) != build_fingerprint(b)
def test_reordering_entity_tables_does_not_change_fingerprint():
a = {"tables": [{"name": "A"}, {"name": "B"}]}
b = {"tables": [{"name": "B"}, {"name": "A"}]}
assert build_fingerprint(a) == build_fingerprint(b)
def test_reordering_catalog_rows_for_fields_and_entities_does_not_change_fingerprint():
a = {
"fields": [{"relation": "A", "field": "ID", "position": 0}, {"relation": "A", "field": "NAME", "position": 1}],
"relations": [{"name": "A"}, {"name": "B"}],
"indexes": [{"name": "IX_A", "segments": ["ID", "NAME"]}, {"name": "IX_B", "segments": ["CODE"]}],
}
b = {
"fields": list(reversed(a["fields"])),
"relations": list(reversed(a["relations"])),
"indexes": list(reversed(a["indexes"])),
}
assert build_fingerprint(a) == build_fingerprint(b)
def test_position_changes_alter_fingerprint_while_segment_position_is_preserved():
fields = {"fields": [{"relation": "A", "field": "ID", "position": 0}, {"relation": "A", "field": "NAME", "position": 1}]}
shifted = {"fields": [{"relation": "A", "field": "ID", "position": 1}, {"relation": "A", "field": "NAME", "position": 0}]}
segments = {"indexes": [{"name": "IX", "segments": [{"field": "A", "position": 0}, {"field": "B", "position": 1}]}]}
reversed_segments = {"indexes": [{"name": "IX", "segments": list(reversed(segments["indexes"][0]["segments"]))}]}
assert build_fingerprint(fields) != build_fingerprint(shifted)
assert build_fingerprint(segments) != build_fingerprint(reversed_segments)
def test_set_values_have_deterministic_order():
assert build_fingerprint({"relations": {"A", "B"}}) == build_fingerprint({"relations": {"B", "A"}})
def test_column_position_and_type_changes_are_relevant_and_order_is_preserved():
base = {"tables": [{"name": "A", "columns": [{"name": "ID", "position": 0, "type": "INTEGER"}, {"name": "NCM", "position": 1, "type": "VARCHAR"}]}]}
reordered = {"tables": [{"name": "A", "columns": list(reversed(base["tables"][0]["columns"]))}]}
changed_type = {"tables": [{"name": "A", "columns": [{"name": "ID", "position": 0, "type": "BIGINT"}, {"name": "NCM", "position": 1, "type": "VARCHAR"}]}]}
assert build_fingerprint(base) != build_fingerprint(reordered)
assert build_fingerprint(base) != build_fingerprint(changed_type)
def test_foreign_key_and_index_changes_are_relevant_but_segments_keep_order():
base = {"foreign_keys": [{"name": "FK", "target": "CUSTOMER", "segments": ["CUSTOMER_ID", "BRANCH_ID"]}], "indexes": [{"name": "IX", "unique": True, "segments": ["A", "B"]}]}
reversed_segments = {"foreign_keys": [{"name": "FK", "target": "CUSTOMER", "segments": ["BRANCH_ID", "CUSTOMER_ID"]}], "indexes": [{"name": "IX", "unique": False, "segments": ["A", "B"]}]}
assert build_fingerprint(base) != build_fingerprint(reversed_segments)
def test_blob_like_and_unknown_large_values_are_redacted():
class UnknownValue:
def __str__(self):
return "secret raw content"
connection = RedactionCatalogReader([[{"SOURCE": b"raw blob", "DESCRIPTION": "x" * 2048, "UNKNOWN": UnknownValue()}]])
metadata = collect_metadata(connection)
values = str(metadata)
assert "raw blob" not in values
assert "secret raw content" not in values
assert "<redacted>" in values
def test_fingerprint_redacts_routine_source_and_default_values():
metadata = {
"engine": {"dialect": 3, "PROCEDURE_SOURCE": "do not store"},
"DEFAULT_SOURCE": "secret",
}
assert build_fingerprint(metadata) == build_fingerprint({"engine": {"dialect": 3}})
def test_tuple_db_api_rows_and_embedded_secret_strings_are_redacted():
value = _safe_value(("safe-name", "password=do-not-store", "raw source text"))
assert value[0] == "safe-name"
assert value[1:] == ["<redacted>", "<redacted>"]
assert "do-not-store" not in str(value)
assert "raw source text" not in str(value)
def test_metadata_queries_never_fetch_source_or_default_blob_contents():
forbidden = (
"DEFAULT_SOURCE",
"TRIGGER_SOURCE",
"PROCEDURE_SOURCE",
"PACKAGE_HEADER_SOURCE",
"PACKAGE_BODY_SOURCE",
)
for query in _CATALOG_QUERIES.values():
assert not any(marker in query for marker in forbidden)
assert "RDB$DEFAULT_SOURCE" not in query
assert "RDB$TRIGGER_SOURCE" not in query
assert "RDB$PROCEDURE_SOURCE" not in query
assert "RDB$PACKAGE_HEADER_SOURCE" not in query
assert "RDB$PACKAGE_BODY_SOURCE" not in query
def test_collection_retains_optional_and_real_errors_and_is_incomplete():
connection = ErrorCatalogReader()
metadata = collect_metadata(connection)
fingerprint = build_fingerprint({"database": []})
assert {error["kind"] for error in metadata["errors"]} == {"optional_unsupported", "query_error"}
assert metadata["errors"] == sorted(
metadata["errors"], key=lambda error: json.dumps(error, sort_keys=True, separators=(",", ":"))
)
assert metadata["incomplete"] is True
assert classify_schema(metadata, {fingerprint}) == SCHEMA_UNKNOWN
def test_classification_distinguishes_supported_changed_and_unknown():
metadata = {"tables": [{"name": "A", "columns": ["ID"]}]}
fingerprint = build_fingerprint(metadata)
assert classify_schema(metadata, {fingerprint}) == SCHEMA_SUPPORTED
assert classify_schema(metadata, {"0" * 64}) == SCHEMA_CHANGED
assert classify_schema(metadata, {fingerprint}, mapping_ambiguous=True) == SCHEMA_UNKNOWN
def test_classification_returns_unknown_for_invalid_allowlists():
metadata = {"tables": [{"name": "A"}]}
for allowlist in (None, 42, [{"not": "hashable"}], (item for item in [["not", "hashable"]])):
assert classify_schema(metadata, allowlist) == SCHEMA_UNKNOWN
class FakeConnection:
def __init__(self):
self.queries = []
def execute(self, sql, params=()):
self.queries.append((sql, params))
return [{"RDB$RELATION_NAME": "PRODUCT", "PASSWORD": "do-not-store", "DATA": b"blob"}]
class RedactionConnection(FakeConnection):
def __init__(self, results):
super().__init__()
self.results = iter(results)
def execute(self, sql, params=()):
self.queries.append((sql, params))
return next(self.results)
class _CatalogReaderMixin:
def _read(self, name):
return self._rows(name)
def list_database(self): return self._read("database")
def list_relations(self): return self._read("relations")
def list_relation_fields(self): return self._read("fields")
def list_keys(self): return self._read("keys")
def list_indexes(self): return self._read("indexes")
def list_foreign_keys(self): return self._read("foreign_keys")
def list_triggers(self): return self._read("triggers")
def list_procedures(self): return self._read("procedures")
def list_procedure_parameters(self): return self._read("procedure_parameters")
def list_packages(self): return self._read("packages")
class ErrorConnection(FakeConnection):
def execute(self, sql, params=()):
self.queries.append((sql, params))
if "RDB$PACKAGES" in sql:
raise RuntimeError("Table unknown RDB$PACKAGES")
if "RDB$PROCEDURES" in sql:
raise PermissionError("no permission for SELECT")
return []
@_approved_fixture_reader(token=_FIXTURE_REGISTRATION_TOKEN)
class ErrorCatalogReader(_CatalogReaderMixin, ReadOnlyCatalogReader):
def __init__(self):
self.queries = []
def _rows(self, name):
self.queries.append(name)
if name == "packages":
raise RuntimeError("Table unknown RDB$PACKAGES")
if name == "procedures":
raise PermissionError("no permission for SELECT")
return []
@_approved_fixture_reader(token=_FIXTURE_REGISTRATION_TOKEN)
class RedactionCatalogReader(_CatalogReaderMixin, ReadOnlyCatalogReader):
def __init__(self, results):
self.results = iter(results)
def _rows(self, name):
return next(self.results)
@_approved_fixture_reader(token=_FIXTURE_REGISTRATION_TOKEN)
class CatalogReaderFixture(_CatalogReaderMixin, ReadOnlyCatalogReader):
def __init__(self):
self.queries = []
def _rows(self, name):
self.queries.append(name)
return [{"RDB$RELATION_NAME": "PRODUCT", "PASSWORD": "do-not-store", "DATA": b"blob"}]
def test_collect_metadata_uses_predefined_read_only_queries_and_redacts_values():
connection = CatalogReaderFixture()
metadata = collect_metadata(connection)
assert set(metadata) == set(_CATALOG_QUERIES) | {"events", "unavailable", "errors", "incomplete"}
assert all("PASSWORD" not in str(value) for value in metadata.values())
assert all(b"blob" not in str(value).encode() for value in metadata.values())
assert "RDB$REF_CONSTRAINTS R" in _CATALOG_QUERIES["foreign_keys"]
assert "RDB$UPDATE_RULE" in _CATALOG_QUERIES["foreign_keys"]
assert "RDB$TRIGGER_SEQUENCE" in _CATALOG_QUERIES["triggers"]
def test_collect_metadata_rejects_arbitrary_execute_adapter():
with pytest.raises(TypeError, match="ReadOnlyCatalogReader"):
collect_metadata(FakeConnection())
def test_collect_metadata_accepts_only_catalog_reader_interface():
metadata = collect_metadata(CatalogReaderFixture())
assert metadata["errors"] == []
assert metadata["incomplete"] is True
def test_all_metadata_queries_pass_the_read_only_interface_validator():
for query in _CATALOG_QUERIES.values():
validate_read_only(query)
+429
View File
@@ -0,0 +1,429 @@
from pathlib import Path
import pytest
from gti_discovery.firebird_connection import (
AUTHORIZED_DATABASE_PATH,
_FirebirdAdapter,
connect_authorized_read_only,
get_database_identity,
validate_authorized_database_path,
_ADAPTER_CREATION_TOKEN,
_create_firebird_adapter,
)
from gti_discovery.catalog import FirebirdCatalog
from gti_discovery.firebird_queries import _ADAPTER_REGISTRATION_TOKEN, _register_approved_adapter
def test_authorized_path_is_the_project_target():
assert AUTHORIZED_DATABASE_PATH == Path(
r"D:\PROJETOS\GTISISTEMAS\Projeto\TESTESOFICIAIS.FDB"
)
def test_validation_normalizes_case_and_relative_segments(tmp_path):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
# The production target is fixed; this exercises the same normalization
# rules with a temporary target without opening a database.
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
try:
assert validate_authorized_database_path(
tmp_path / "sub" / ".." / "testesoficiais.fdb"
) == target.resolve()
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_validation_rejects_missing_directory_and_alternate_file(tmp_path):
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
authorized = tmp_path / "TESTESOFICIAIS.FDB"
authorized.write_bytes(b"database")
firebird_connection.AUTHORIZED_DATABASE_PATH = authorized
try:
with pytest.raises(ValueError, match="authorized"):
validate_authorized_database_path(tmp_path / "missing.FDB")
alternate = tmp_path / "ERP.FDB"
alternate.write_bytes(b"database")
with pytest.raises(ValueError, match="authorized"):
validate_authorized_database_path(alternate)
authorized.unlink()
with pytest.raises(ValueError, match="regular file"):
validate_authorized_database_path(authorized)
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_rejected_path_never_invokes_driver(tmp_path):
calls = []
class Driver:
def connect(self, **kwargs):
calls.append(kwargs)
raise AssertionError("driver must not be invoked")
with pytest.raises(ValueError):
connect_authorized_read_only(tmp_path / "ERP.FDB", Driver())
assert calls == []
def test_connection_requests_and_proves_read_only_mode(tmp_path):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
calls = []
class DriverConnection:
default_tpb = b"\x03\x08\x06\x0f\x11"
read_only = True
read_only_evidence = {
"accepted": True,
"isolation_level": "fixture-read-committed-ro",
}
def execute(self, sql, params=()):
return [("TESTESOFICIAIS.FDB",)]
class Driver:
def connect(self, **kwargs):
calls.append(kwargs)
return _create_firebird_adapter(DriverConnection(), {
"accepted": True,
"driver_api": "fdb.Connection.default_tpb",
"configured_tpb": b"\x03\x08\x06\x0f\x11",
"observed_default_tpb": b"\x03\x08\x06\x0f\x11",
}, _ADAPTER_CREATION_TOKEN)
try:
connection = connect_authorized_read_only(target, Driver(), "fbclient.dll")
assert isinstance(connection, FirebirdCatalog)
assert connection.read_only is True
assert calls == [
{
"database": str(target.resolve()),
"read_only": True,
"client_library": "fbclient.dll",
}
]
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_connection_fails_closed_without_proven_read_only_mode(tmp_path):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
class Driver:
def connect(self, **kwargs):
return object()
try:
with pytest.raises(ValueError, match="read-only"):
connect_authorized_read_only(target, Driver())
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_connection_rejects_driver_that_ignores_read_only_setting(tmp_path):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
class IgnoringConnection:
read_only = True
def close(self):
self.closed = True
raw_connection = IgnoringConnection()
class Driver:
def connect(self, **kwargs):
return raw_connection
try:
with pytest.raises(ValueError, match="read-only"):
connect_authorized_read_only(target, Driver())
assert raw_connection.closed is True
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_connection_rejects_lookalike_sealed_adapter_from_untrusted_driver(tmp_path):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
class Driver:
def connect(self, **kwargs):
return _FirebirdAdapter(object(), {
"accepted": True,
"driver_api": "fdb.Connection.default_tpb",
"configured_tpb": b"\x03\x08\x06\x0f\x11",
"observed_default_tpb": b"\x03\x08\x06\x0f\x11",
})
try:
with pytest.raises((TypeError, ValueError)):
connect_authorized_read_only(target, Driver())
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
@pytest.mark.parametrize("read_only", [False, None])
def test_connection_closes_raw_connection_without_read_only_proof(tmp_path, read_only):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
class DriverConnection:
def close(self):
self.closed = True
raw_connection = DriverConnection()
if read_only is not None:
raw_connection.read_only = read_only
raw_connection.closed = False
class Driver:
def connect(self, **kwargs):
return raw_connection
try:
with pytest.raises(ValueError, match="read-only"):
connect_authorized_read_only(target, Driver())
assert raw_connection.closed is True
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_connection_closes_malformed_raw_connection(tmp_path):
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
from gti_discovery import firebird_connection
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
class DriverConnection:
read_only = True
def close(self):
self.closed = True
raw_connection = DriverConnection()
raw_connection.closed = False
class Driver:
def connect(self, **kwargs):
return raw_connection
try:
with pytest.raises(ValueError, match="read-only"):
connect_authorized_read_only(target, Driver())
assert raw_connection.closed is True
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
def test_identity_helper_uses_only_database_monitor_query():
from gti_discovery.collector import _CATALOG_QUERIES
calls = []
class Adapter:
read_only = True
def execute(self, sql, params=()):
calls.append((sql, params))
return [("TESTESOFICIAIS.FDB",)]
_register_approved_adapter(Adapter, _ADAPTER_REGISTRATION_TOKEN)
assert get_database_identity(FirebirdCatalog(Adapter())) == "TESTESOFICIAIS.FDB"
assert calls == [(_CATALOG_QUERIES["database"], ())]
def test_driver_fixture_uses_runtime_fdb_read_only_constant(monkeypatch):
import importlib.util
script = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
spec = importlib.util.spec_from_file_location("run_firebird_discovery_fixture", script)
script_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(script_module)
class FdbFixture:
ISOLATION_LEVEL_READ_COMMITED_RO = b"\x03\x08\x06\x0f\x11"
def connect(self, **kwargs):
self.kwargs = kwargs
return type("Connection", (), {
"default_tpb": kwargs["isolation_level"],
"read_only_evidence": {
"accepted": True,
"isolation_level": kwargs["isolation_level"],
}
})()
fdb = FdbFixture()
monkeypatch.setenv("FB_USER", "fixture-user")
monkeypatch.setenv("FB_PASSWORD", "fixture-password")
driver = script_module._FdbDriver(fdb)
driver.connect(database="target.fdb", read_only=True)
assert fdb.kwargs["isolation_level"] is fdb.ISOLATION_LEVEL_READ_COMMITED_RO
def test_driver_rejects_connection_that_ignores_fdb_read_only_tpb(monkeypatch):
import importlib.util
script = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
spec = importlib.util.spec_from_file_location("run_firebird_discovery_ignored_tpb", script)
script_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(script_module)
class FdbFixture:
ISOLATION_LEVEL_READ_COMMITED_RO = b"\x03\x08\x06\x0f\x11"
def connect(self, **kwargs):
self.kwargs = kwargs
return type("Connection", (), {
"default_tpb": b"ordinary-read-write-tpb",
"read_only": True,
"read_only_evidence": {
"accepted": True,
"isolation_level": kwargs["isolation_level"],
},
})()
monkeypatch.setenv("FB_USER", "fixture-user")
monkeypatch.setenv("FB_PASSWORD", "fixture-password")
with pytest.raises(RuntimeError, match="read-only"):
script_module._FdbDriver(FdbFixture()).connect(database="target.fdb", read_only=True)
def test_driver_uses_adapter_configuration_evidence_not_raw_claims(monkeypatch):
import importlib.util
script = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
spec = importlib.util.spec_from_file_location("run_firebird_discovery_claim", script)
script_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(script_module)
class FdbFixture:
ISOLATION_LEVEL_READ_COMMITED_RO = b"\x03\x08\x06\x0f\x11"
def connect(self, **kwargs):
return type("Connection", (), {
"default_tpb": kwargs["isolation_level"],
"read_only": False,
"read_only_evidence": {"accepted": True, "isolation_level": kwargs["isolation_level"]},
})()
monkeypatch.setenv("FB_USER", "fixture-user")
monkeypatch.setenv("FB_PASSWORD", "fixture-password")
adapter = script_module._FdbDriver(FdbFixture()).connect(database="target.fdb", read_only=True)
assert adapter.read_only_evidence["configured_tpb"] == b"\x03\x08\x06\x0f\x11"
assert adapter.read_only_evidence["observed_default_tpb"] == b"\x03\x08\x06\x0f\x11"
def test_installed_fdb_runtime_exposes_read_only_isolation_api_without_connecting():
import fdb
import importlib.util
script = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
spec = importlib.util.spec_from_file_location("run_firebird_discovery_runtime", script)
script_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(script_module)
assert script_module._read_only_isolation_level(fdb) is getattr(
fdb, "ISOLATION_LEVEL_READ_COMMITED_RO"
)
def test_identity_mismatch_is_rejected_after_connection(tmp_path, monkeypatch):
from gti_discovery import firebird_connection
target = tmp_path / "TESTESOFICIAIS.FDB"
target.write_bytes(b"database")
original = firebird_connection.AUTHORIZED_DATABASE_PATH
firebird_connection.AUTHORIZED_DATABASE_PATH = target
import importlib.util
script = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
spec = importlib.util.spec_from_file_location("run_firebird_discovery_identity", script)
script_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(script_module)
monkeypatch.setenv("FB_USER", "fixture-user")
monkeypatch.setenv("FB_PASSWORD", "fixture-password")
monkeypatch.setattr(script_module, "_client_metadata", lambda path: {
"path": str(path), "exists": True, "version": "2.5.9.27139"
})
monkeypatch.setattr(script_module, "_security_gate", lambda path, driver: {
"path": str(path), "exists": True, "version": "2.5.9.27139"
})
class RawConnection:
default_tpb = b"\x03\x08\x06\x0f\x11"
read_only = True
read_only_evidence = {
"accepted": True,
"isolation_level": "fixture-read-committed-ro",
}
def execute(self, sql, params=()):
return [("OTHER.FDB",)]
def close(self):
pass
_register_approved_adapter(RawConnection, _ADAPTER_REGISTRATION_TOKEN)
class Driver:
version = "test-driver"
def compatibility_evidence(self, client_version):
return {
"client_version": client_version,
"fdb_version": "2.0.4",
"firebird_version": "2.5.9",
"read_only_isolation": True,
"read_only_api": "fdb.Connection.default_tpb",
"read_only_tpb": b"\x03\x08\x06\x0f\x11",
}
def connect(self, **kwargs):
return _create_firebird_adapter(RawConnection(), {
"accepted": True,
"driver_api": "fdb.Connection.default_tpb",
"configured_tpb": b"\x03\x08\x06\x0f\x11",
"observed_default_tpb": b"\x03\x08\x06\x0f\x11",
}, _ADAPTER_CREATION_TOKEN)
try:
result = script_module.run_discovery(target, driver=Driver())
assert result["status"] == "MISSION_00_BLOCKED"
assert result["blocker"] == "identity_mismatch"
assert result["connection_attempts"] == 1
assert result["other_databases_accessed"] == "UNKNOWN"
assert result["application_writes"] == 0
finally:
firebird_connection.AUTHORIZED_DATABASE_PATH = original
+762
View File
@@ -0,0 +1,762 @@
import importlib.util
import importlib.metadata
import json
from pathlib import Path
import pytest
SCRIPT = Path(__file__).parents[1] / "scripts" / "run_firebird_discovery.py"
def load_script():
spec = importlib.util.spec_from_file_location("run_firebird_discovery", SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_alternate_target_is_rejected_before_driver_invocation():
module = load_script()
calls = []
class Driver:
def connect(self, **kwargs):
calls.append(kwargs)
raise AssertionError("driver must not be invoked")
with pytest.raises(ValueError, match="authorized"):
module.run_discovery(Path("D:/other.FDB"), driver=Driver())
assert calls == []
def test_first_sql_is_exact_identity_query(monkeypatch):
from gti_discovery.collector import _CATALOG_QUERIES
from gti_discovery.firebird_connection import _ADAPTER_CREATION_TOKEN, _create_firebird_adapter
from gti_discovery.firebird_queries import _ADAPTER_REGISTRATION_TOKEN, _register_approved_adapter
module = load_script()
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
module._client_metadata = lambda path: {'path': str(path), 'exists': True, 'version': '2.5.9.27139'}
monkeypatch.setattr(module, '_security_gate', lambda path, driver: {'path': str(path), 'exists': True, 'version': '2.5.9.27139'})
module.os.environ['FB_USER'] = 'fixture-user'
module.os.environ['FB_PASSWORD'] = 'fixture-password'
module._client_metadata = lambda path: {'path': str(path), 'exists': True, 'version': '2.5.9.27139'}
calls = []
class RawConnection:
default_tpb = b"\x03\x08\x06\x0f\x11"
read_only = True
read_only_evidence = {
"accepted": True,
"isolation_level": "fixture-read-committed-ro",
}
def execute(self, sql, params=()):
calls.append((sql, params))
return [("TESTESOFICIAIS.FDB",)]
def cursor(self):
return self
def fetchall(self):
return [("TESTESOFICIAIS.FDB",)]
def close(self):
pass
_register_approved_adapter(RawConnection, _ADAPTER_REGISTRATION_TOKEN)
class Driver:
version = "test-driver"
def compatibility_evidence(self, client_version):
return {
"client_version": client_version,
"fdb_version": "2.0.4",
"firebird_version": "2.5.9",
"read_only_isolation": True,
"read_only_api": "fdb.Connection.default_tpb",
"read_only_tpb": b"\x03\x08\x06\x0f\x11",
}
def connect(self, **kwargs):
return _create_firebird_adapter(RawConnection(), {
"accepted": True,
"driver_api": "fdb.Connection.default_tpb",
"configured_tpb": b"\x03\x08\x06\x0f\x11",
"observed_default_tpb": b"\x03\x08\x06\x0f\x11",
}, _ADAPTER_CREATION_TOKEN)
result = module.run_discovery(module.AUTHORIZED_DATABASE_PATH, driver=Driver())
assert calls[0] == (_CATALOG_QUERIES["database"], ())
assert result["status"] == "MISSION_00_BLOCKED"
def test_direct_run_requires_security_gate_before_driver(monkeypatch, tmp_path):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'AUTHORIZED_DATABASE_PATH', target)
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.delenv('FB_USER', raising=False)
monkeypatch.delenv('FB_PASSWORD', raising=False)
calls = []
class Driver:
version = '1.0'
def connect(self, **kwargs):
calls.append(kwargs)
raise AssertionError('driver must not be invoked')
result = module.run_discovery(target, driver=Driver())
assert result == {
'status': 'MISSION_00_BLOCKED',
'blocker': 'credentials_unavailable',
'application_writes': 0,
'other_databases_accessed': 0,
'connection_attempts': 0,
}
assert calls == []
def test_direct_run_blocks_unproven_firebird_client_compatibility(monkeypatch, tmp_path):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'AUTHORIZED_DATABASE_PATH', target)
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
monkeypatch.setattr(module, '_client_metadata', lambda path: {
'path': str(path), 'exists': True, 'version': '2.6.0'
})
calls = []
class Driver:
version = '1.0'
def connect(self, **kwargs):
calls.append(kwargs)
raise AssertionError('driver must not be invoked')
result = module.run_discovery(target, driver=Driver())
assert result['status'] == 'MISSION_00_BLOCKED'
assert result['blocker'] == 'compatibility_unproven'
assert calls == []
def test_direct_run_completes_sanitized_discovery_workflow(monkeypatch, tmp_path):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'AUTHORIZED_DATABASE_PATH', target)
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
monkeypatch.setattr(module, '_client_metadata', lambda path: {
'path': str(path), 'exists': True, 'version': '2.5.9.27139'
})
monkeypatch.setattr(module, '_read_client_architecture', lambda path: 'x64')
monkeypatch.setattr(module, '_security_gate', lambda path, driver: {'path': str(path), 'exists': True, 'version': '2.5.9.27139'})
calls = []
class Connection:
def close(self):
calls.append('close')
def sample_relation_rows(self, relation, columns, limit):
calls.append(('sample', relation, tuple(columns), limit))
return [{'ID': 1, 'DESCRIPTION': 'widget'}]
class Driver:
version = '1.0'
def compatibility_evidence(self, client_version):
return {
'client_version': client_version,
'fdb_version': '2.0.4',
'firebird_version': '2.5.9',
'read_only_isolation': True,
'read_only_api': 'fdb.Connection.default_tpb',
'read_only_tpb': b'\x03\x08\x06\x0f\x11',
}
def connect(self, **kwargs):
calls.append('connect')
return Connection()
def facts(self):
calls.append('facts')
return {'engine_version': '2.5.9', 'database_sql_dialect': 3}
monkeypatch.setattr(module, 'connect_authorized_read_only', lambda path, driver, client_library: Connection())
monkeypatch.setattr(module, 'get_database_identity', lambda connection: 'TESTESOFICIAIS.FDB')
metadata = {
'database': [{'MON$DATABASE_NAME': 'TESTESOFICIAIS.FDB'}],
'relations': [{'RELATION_NAME': 'PRODUCTS'}],
'fields': [
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'ID', 'TYPE_NAME': 'INTEGER'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'DESCRIPTION', 'TYPE_NAME': 'VARCHAR'},
],
'keys': [{'RELATION_NAME': 'PRODUCTS', 'CONSTRAINT_TYPE': 'PRIMARY KEY', 'FIELD_NAME': 'ID'}],
'foreign_keys': [], 'indexes': [], 'triggers': [], 'procedures': [],
'procedure_parameters': [], 'packages': [], 'generators': [],
'dependencies': [], 'monitoring': [], 'events': [],
'unavailable': ['events', 'packages', 'procedure_parameters'],
'errors': [
{'kind': 'optional_unsupported', 'query': 'packages'},
{'kind': 'optional_unsupported', 'query': 'procedure_parameters'},
{'kind': 'query_error', 'query': 'monitoring'},
], 'incomplete': True,
}
monkeypatch.setattr(module, 'collect_metadata', lambda reader: (calls.append('metadata') or metadata))
valid_mapping = {
'product': {'classification': 'INFERRED', 'relation': None, 'evidence': {}, 'candidates': []}, 'fields': {},
'fiscal': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'stock': {'classification': 'UNKNOWN', 'relations': {}},
'media': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'incremental': {'classification': 'UNKNOWN', 'evidence': {'timestamps': [], 'triggers': 0, 'generators': 0}}, 'triggers': [],
}
monkeypatch.setattr(module, 'build_schema_mapping', lambda data, samples: (calls.append('mapping') or valid_mapping))
monkeypatch.setattr(module, 'build_fingerprint', lambda data: (calls.append('fingerprint') or 'a' * 64))
monkeypatch.setattr(module, 'classify_schema', lambda data, expected, mapping_ambiguous=False: (calls.append('classify') or 'SCHEMA_UNKNOWN'))
result = module.run_discovery(target, driver=Driver())
assert result['status'] == 'CONNECTED'
assert result['schema_mapping'] == valid_mapping
assert result['fingerprint'] == 'a' * 64
assert result['schema_classification'] == 'SCHEMA_UNKNOWN'
assert result['application_writes'] == 0
assert result['other_databases_accessed'] == 0
assert result['connection_attempts'] == 1
assert 'database_name' not in result
output = json.dumps(result, sort_keys=True)
assert 'client_library' not in result
assert 'client_version' not in result
assert 'compatibility_evidence' not in result
assert str(target) not in output
assert ('sample', 'PRODUCTS', ('ID', 'DESCRIPTION'), 5) in calls
assert calls.index('metadata') < calls.index('mapping') < calls.index('fingerprint') < calls.index('classify')
def test_direct_run_never_reports_connected_without_complete_evidence(monkeypatch, tmp_path):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
monkeypatch.setattr(module, '_client_metadata', lambda path: {
'path': str(path), 'exists': True, 'version': '2.5.9.27139'
})
monkeypatch.setattr(module, '_security_gate', lambda path, driver: {'path': str(path), 'exists': True, 'version': '2.5.9.27139'})
class Connection:
def close(self):
pass
def sample_relation_rows(self, relation, columns, limit):
return [{'ID': 1}]
class Driver:
version = '1.0'
def compatibility_evidence(self, client_version):
return {
'client_version': client_version,
'fdb_version': '2.0.4',
'firebird_version': '2.5.9',
'read_only_isolation': True,
'read_only_api': 'fdb.Connection.default_tpb',
'read_only_tpb': b'\x03\x08\x06\x0f\x11',
}
def facts(self):
return {}
monkeypatch.setattr(module, 'connect_authorized_read_only', lambda path, driver, client_library: Connection())
monkeypatch.setattr(module, 'get_database_identity', lambda connection: 'TESTESOFICIAIS.FDB')
monkeypatch.setattr(module, 'collect_metadata', lambda reader: {
'relations': [{'RELATION_NAME': 'PRODUCTS'}],
'fields': [{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'ID'}],
'errors': [], 'unavailable': [], 'incomplete': True,
})
monkeypatch.setattr(module, 'build_schema_mapping', lambda metadata, samples: {
'product': {'classification': 'UNKNOWN', 'relation': None, 'evidence': {}, 'candidates': []}, 'fields': {},
'fiscal': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'stock': {'classification': 'UNKNOWN', 'relations': {}},
'media': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'incremental': {'classification': 'UNKNOWN', 'evidence': {'timestamps': [], 'triggers': 0, 'generators': 0}}, 'triggers': [],
})
result = module.run_discovery(target, driver=Driver())
assert result['status'] == 'SCHEMA_UNKNOWN'
assert result['fingerprint'] is None
def test_identity_mismatch_reports_unknown_unauthorized_access_counter(monkeypatch, tmp_path):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
monkeypatch.setattr(module, '_security_gate', lambda path, driver: {'client': {}})
class Connection:
def close(self):
pass
monkeypatch.setattr(module, 'connect_authorized_read_only', lambda path, driver, client_library: Connection())
monkeypatch.setattr(module, 'get_database_identity', lambda connection: 'OTHER.FDB')
result = module.run_discovery(target, driver=object())
assert result == {
'status': 'MISSION_00_BLOCKED',
'blocker': 'identity_mismatch',
'application_writes': 0,
'other_databases_accessed': 'UNKNOWN',
'connection_attempts': 1,
}
def test_serialized_result_removes_database_identity_and_sensitive_evidence(monkeypatch, tmp_path):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
monkeypatch.setattr(module, '_security_gate', lambda path, driver: {'client': {}})
class Connection:
def close(self):
pass
def sample_relation_rows(self, relation, columns, limit):
return [{'ID': 1, 'DESCRIPTION': 'widget'}]
monkeypatch.setattr(module, 'connect_authorized_read_only', lambda path, driver, client_library: Connection())
monkeypatch.setattr(module, 'get_database_identity', lambda connection: 'TESTESOFICIAIS.FDB')
monkeypatch.setattr(module, 'collect_metadata', lambda reader: {
'database': [{'MON$DATABASE_NAME': str(target), 'PASSWORD': 'do-not-leak'}],
'relations': [{'RELATION_NAME': 'PRODUCTS'}],
'fields': [{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'ID'}, {'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'DESCRIPTION'}],
'keys': [], 'foreign_keys': [], 'indexes': [], 'triggers': [], 'procedures': [],
'procedure_parameters': [], 'packages': [], 'generators': [], 'dependencies': [],
'monitoring': [], 'events': [], 'unavailable': [], 'errors': [], 'incomplete': False,
})
valid_mapping = {
'product': {'classification': 'INFERRED', 'relation': None, 'evidence': {}, 'candidates': []}, 'fields': {},
'fiscal': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'stock': {'classification': 'UNKNOWN', 'relations': {}}, 'media': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'incremental': {'classification': 'UNKNOWN', 'evidence': {'timestamps': [], 'triggers': 0, 'generators': 0}}, 'triggers': [],
}
monkeypatch.setattr(module, 'build_schema_mapping', lambda data, samples: valid_mapping)
result = module.run_discovery(target, driver=object())
serialized = json.dumps(result, sort_keys=True)
assert result['status'] == 'CONNECTED'
assert str(target) not in serialized
assert 'PASSWORD' not in serialized
assert 'client_library' not in serialized
def test_sample_plan_excludes_denylisted_fields_before_query():
module = load_script()
metadata = {
'relations': [{'RELATION_NAME': 'PRODUCTS'}],
'fields': [
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'ID', 'TYPE_NAME': 'INTEGER'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'DESCRIPTION', 'TYPE_NAME': 'VARCHAR'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'CUSTOMER_EMAIL', 'TYPE_NAME': 'VARCHAR'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'UNIT_PRICE', 'TYPE_NAME': 'DECIMAL'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'TAX_RATE', 'TYPE_NAME': 'DECIMAL'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'IMAGE_BLOB', 'TYPE_NAME': 'BLOB'},
],
'keys': [{'RELATION_NAME': 'PRODUCTS', 'CONSTRAINT_TYPE': 'PRIMARY KEY', 'FIELD_NAME': 'ID'}],
}
calls = []
class Reader:
def sample_relation_rows(self, relation, columns, limit):
calls.append((relation, tuple(columns), limit))
return [{'ID': 1, 'DESCRIPTION': 'widget'}]
module._collect_samples(Reader(), metadata)
assert calls == [('PRODUCTS', ('ID', 'DESCRIPTION'), 5)]
def test_invalid_sanitized_sample_plan_is_rejected_before_query(monkeypatch):
module = load_script()
monkeypatch.setattr(module, '_permitted_sample_plan', lambda metadata: [
('PRODUCTS', ['CUSTOMER_EMAIL'])
])
class Reader:
def sample_relation_rows(self, relation, columns, limit):
raise AssertionError('invalid plan must not reach reader')
with pytest.raises(RuntimeError, match='sample plan'):
module._collect_samples(Reader(), {'relations': [], 'fields': []})
@pytest.mark.parametrize('bad_kind', ['metadata', 'mapping', 'fingerprint', 'classification'])
def test_malformed_success_evidence_never_reports_connected(monkeypatch, tmp_path, bad_kind):
module = load_script()
target = tmp_path / 'TESTESOFICIAIS.FDB'
target.write_bytes(b'database')
monkeypatch.setattr(module, 'validate_authorized_database_path', lambda path: target)
monkeypatch.setenv('FB_USER', 'fixture-user')
monkeypatch.setenv('FB_PASSWORD', 'fixture-password')
monkeypatch.setattr(module, '_client_metadata', lambda path: {
'path': str(path), 'exists': True, 'version': '2.5.9.27139'
})
class Connection:
def close(self):
pass
def sample_relation_rows(self, relation, columns, limit):
return [{'ID': 1, 'DESCRIPTION': 'widget'}]
class Driver:
version = '1.0'
def compatibility_evidence(self, client_version):
return {
'client_version': client_version,
'fdb_version': '2.0.4',
'firebird_version': '2.5.9',
'read_only_isolation': True,
'read_only_api': 'fdb.Connection.default_tpb',
'read_only_tpb': b'\x03\x08\x06\x0f\x11',
}
def facts(self):
return {}
metadata = {
'relations': [{'RELATION_NAME': 'PRODUCTS'}],
'fields': [
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'ID'},
{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'DESCRIPTION'},
],
'keys': [{'RELATION_NAME': 'PRODUCTS', 'CONSTRAINT_TYPE': 'PRIMARY KEY', 'FIELD_NAME': 'ID'}],
'errors': [], 'unavailable': [], 'incomplete': False,
}
monkeypatch.setattr(module, 'connect_authorized_read_only', lambda path, driver, client_library: Connection())
monkeypatch.setattr(module, 'get_database_identity', lambda connection: 'TESTESOFICIAIS.FDB')
monkeypatch.setattr(module, 'collect_metadata', lambda reader: None if bad_kind == 'metadata' else metadata)
valid_mapping = {
'product': {'classification': 'INFERRED', 'candidates': []}, 'fields': {},
'fiscal': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'stock': {'classification': 'UNKNOWN', 'relations': {}},
'media': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'incremental': {'classification': 'UNKNOWN'}, 'triggers': [],
}
monkeypatch.setattr(module, 'build_schema_mapping', lambda data, samples: None if bad_kind == 'mapping' else valid_mapping)
monkeypatch.setattr(module, 'build_fingerprint', lambda data: 'invalid' if bad_kind == 'fingerprint' else 'a' * 64)
monkeypatch.setattr(module, 'classify_schema', lambda data, expected, mapping_ambiguous=False: 'INVALID' if bad_kind == 'classification' else 'SCHEMA_UNKNOWN')
result = module.run_discovery(target, driver=Driver())
assert result['status'] in {'MISSION_00_BLOCKED', 'SCHEMA_UNKNOWN'}
assert result['status'] != 'CONNECTED'
def test_runtime_compatibility_requires_matching_concrete_evidence(monkeypatch):
module = load_script()
monkeypatch.setattr(module, '_client_metadata', lambda path: {
'path': str(path), 'exists': True, 'version': '2.5.9.27139'
})
class Driver:
version = '1.0'
def compatibility_evidence(self, client_version):
return {
'client_version': client_version,
'fdb_version': '2.0.4',
'firebird_version': '2.6.0',
'read_only_isolation': True,
'read_only_api': 'fdb.Connection.default_tpb',
'read_only_tpb': b'\x03\x08\x06\x0f\x11',
}
with pytest.raises(RuntimeError, match='compatib'):
module._validate_runtime_compatibility(Driver())
def test_runtime_compatibility_rejects_injected_fdb_module(monkeypatch):
module = load_script()
monkeypatch.setattr(module, "_client_metadata", lambda path: {
"path": str(path), "exists": True, "version": "2.5.9.27139"
})
class FakeFdb:
__version__ = "2.0.4"
ISOLATION_LEVEL_READ_COMMITED_RO = b"\x03\x08\x06\x0f\x11"
driver = module._FdbDriver(FakeFdb())
with pytest.raises(RuntimeError, match="compatib"):
module._validate_runtime_compatibility(driver)
def test_driver_rejects_unexpected_connection_settings(monkeypatch):
module = load_script()
monkeypatch.setenv("FB_USER", "fixture-user")
monkeypatch.setenv("FB_PASSWORD", "fixture-password")
class FdbFixture:
ISOLATION_LEVEL_READ_COMMITED_RO = b"\x03\x08\x06\x0f\x11"
def connect(self, **kwargs):
return type("Connection", (), {"default_tpb": kwargs["isolation_level"]})()
driver = module._FdbDriver(FdbFixture())
with pytest.raises(RuntimeError, match="settings"):
driver.connect(database="target.fdb", read_only=True, ignored=True)
def test_preimport_compatibility_rejects_wrong_fdb_distribution(monkeypatch):
module = load_script()
monkeypatch.setattr(module, '_client_metadata', lambda path: {
'path': str(path), 'exists': True, 'version': '2.5.9.27139'
})
monkeypatch.setattr(module.importlib.metadata, 'version', lambda name: '2.0.3')
with pytest.raises(RuntimeError, match='compatib'):
module._validate_preimport_compatibility()
def test_preimport_gate_checks_provenance_before_distribution(monkeypatch):
module = load_script()
calls = []
monkeypatch.setattr(module, "validate_fdb_provenance", lambda: calls.append("provenance") or {"status": "VALID"})
monkeypatch.setattr(module, "_client_metadata", lambda path: calls.append("client") or {
"path": str(path), "exists": True, "version": "2.5.9.27139"
})
monkeypatch.setattr(module, "_read_client_architecture", lambda path: calls.append("architecture") or "x64")
monkeypatch.setattr(module.importlib.metadata, "version", lambda name: calls.append("distribution") or "2.0.4")
module._validate_preimport_compatibility()
assert calls == ["provenance", "client", "architecture", "distribution"]
def test_main_sanitizes_provenance_when_blocked(monkeypatch, capsys):
module = load_script()
evidence = {
"status": "UNKNOWN", "module_file": r"C:\secret\fdb\__init__.py",
"distribution_location": r"C:\secret\site-packages", "shadowing": [],
"errors": ["ImportError"],
}
monkeypatch.setenv("FB_USER", "user")
monkeypatch.setenv("FB_PASSWORD", "password")
monkeypatch.setattr(module, "validate_fdb_provenance", lambda: (_ for _ in ()).throw(
module.ProvenanceError("fdb provenance is unknown", evidence)
))
assert module.main([]) == 2
output = json.loads(capsys.readouterr().out)
assert output["status"] == "MISSION_00_BLOCKED"
assert output["blocker"] == "compatibility_unproven"
assert "fdb_provenance" not in output
assert r"C:\secret" not in json.dumps(output)
def test_valid_mapping_requires_nested_classifications():
module = load_script()
mapping = {
'product': {'classification': 'INFERRED', 'candidates': []},
'fields': {'DESCRIPTION': {}},
'fiscal': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'stock': {'classification': 'UNKNOWN', 'relations': {}},
'media': {'classification': 'UNKNOWN', 'fields': {}, 'relations': {}},
'incremental': {'classification': 'UNKNOWN'},
'triggers': [],
}
assert not module._valid_mapping(mapping)
def test_valid_mapping_accepts_mapping_produced_by_builder():
module = load_script()
from gti_discovery.mapping import build_schema_mapping
metadata = {
"relations": [{"name": "PRODUCT"}],
"fields": [
{"relation": "PRODUCT", "name": "ID", "type": "INTEGER"},
{"relation": "PRODUCT", "name": "DESCRIPTION", "type": "VARCHAR"},
],
"keys": [{"relation": "PRODUCT", "constraint_type": "PRIMARY KEY", "fields": ["ID"]}],
}
mapping = build_schema_mapping(metadata, {"PRODUCT": [{"ID": 1, "DESCRIPTION": "masked"}]})
assert module._valid_mapping(mapping)
def test_valid_metadata_accepts_catalog_error_records_and_unavailable_names():
module = load_script()
metadata = {
"relations": [],
"fields": [],
"errors": [{"kind": "optional_unsupported", "query": "packages"}],
"unavailable": ["packages"],
}
assert module._valid_metadata(metadata)
def test_valid_mapping_rejects_candidate_with_parent_only_fields():
module = load_script()
from gti_discovery.mapping import build_schema_mapping
metadata = {
"relations": [{"name": "PRODUCT"}],
"fields": [
{"relation": "PRODUCT", "name": "ID", "type": "INTEGER"},
{"relation": "PRODUCT", "name": "DESCRIPTION", "type": "VARCHAR"},
],
"keys": [{"relation": "PRODUCT", "constraint_type": "PRIMARY KEY", "fields": ["ID"]}],
}
mapping = build_schema_mapping(metadata, {"PRODUCT": [{"ID": 1, "DESCRIPTION": "masked"}]})
mapping["product"]["candidates"][0]["candidates"] = []
assert not module._valid_mapping(mapping)
def test_valid_mapping_rejects_deep_malformed_evidence_and_relationships():
module = load_script()
mapping = {
"product": {
"classification": "INFERRED", "relation": "PRODUCTS",
"evidence": {"primary_key": [], "field_count": 1, "semantic_fields": []},
"candidates": [{
"relation": "PRODUCTS", "classification": "INFERRED",
"evidence": {"primary_key": [], "field_count": 1, "semantic_fields": []},
}],
},
"fields": {},
"fiscal": {"classification": "UNKNOWN", "fields": {}, "relations": {}},
"stock": {"classification": "UNKNOWN", "relations": {}},
"media": {"classification": "UNKNOWN", "fields": {}, "relations": {}},
"incremental": {"classification": "UNKNOWN", "evidence": {"timestamps": [], "triggers": 0, "generators": 0}},
"triggers": [],
}
mapping["stock"]["relations"]["STOCK"] = {
"classification": "INFERRED",
"evidence": "relation name",
"fields": {"ID": {"classification": "INFERRED", "type": "INTEGER"}},
"relationships": [{
"child_relation": "STOCK", "child_field": "PRODUCT_ID",
"referenced_relation": "PRODUCTS", "referenced_field": 7,
}],
}
assert not module._valid_mapping(mapping)
@pytest.mark.parametrize("mutate", [
lambda mapping: mapping["product"]["candidates"].append({"classification": "UNKNOWN"}),
lambda mapping: mapping["fields"].update({"DESCRIPTION": {"classification": "UNKNOWN", "sample_present": "yes"}}),
lambda mapping: mapping["fiscal"]["relations"].update({"TAX": {"classification": "UNKNOWN", "fields": []}}),
lambda mapping: mapping["stock"]["relations"].update({"STOCK": {"classification": "UNKNOWN", "evidence": 4, "fields": {}, "relationships": []}}),
lambda mapping: mapping["triggers"].append({"classification": "UNKNOWN", "relation": "PRODUCTS"}),
])
def test_valid_mapping_rejects_malformed_nested_records(mutate):
module = load_script()
mapping = {
"product": {"classification": "INFERRED", "relation": "PRODUCTS", "evidence": {}, "candidates": []},
"fields": {},
"fiscal": {"classification": "UNKNOWN", "fields": {}, "relations": {}},
"stock": {"classification": "UNKNOWN", "relations": {}},
"media": {"classification": "UNKNOWN", "fields": {}, "relations": {}},
"incremental": {"classification": "UNKNOWN", "evidence": {"timestamps": [], "triggers": 0, "generators": 0}},
"triggers": [],
}
mutate(mapping)
assert not module._valid_mapping(mapping)
def test_cli_target_is_fixed_to_authorized_database():
module = load_script()
assert module.DEFAULT_TARGET == module.AUTHORIZED_DATABASE_PATH
def test_client_metadata_uses_runtime_version_and_path(monkeypatch, tmp_path):
module = load_script()
client = tmp_path / "fbclient.dll"
client.write_bytes(b"client")
monkeypatch.setattr(module, "_read_file_version", lambda path: "2.5.9.27139")
assert module._client_metadata(client) == {
"path": str(client),
"exists": True,
"version": "2.5.9.27139",
}
@pytest.mark.parametrize("missing", ["FB_USER", "FB_PASSWORD"])
def test_missing_credentials_block_before_driver_invocation(monkeypatch, capsys, missing):
module = load_script()
monkeypatch.delenv("FB_USER", raising=False)
monkeypatch.delenv("FB_PASSWORD", raising=False)
monkeypatch.setenv("FB_PASSWORD" if missing == "FB_USER" else "FB_USER", "present")
calls = []
class Driver:
def connect(self, **kwargs):
calls.append(kwargs)
raise AssertionError("driver must not be invoked")
monkeypatch.setattr(module, "_FdbDriver", lambda fdb: Driver())
assert module.main([]) == 2
output = json.loads(capsys.readouterr().out)
assert output == {
"application_writes": 0,
"blocker": "credentials_unavailable",
"other_databases_accessed": 0,
"connection_attempts": 0,
"status": "MISSION_00_BLOCKED",
}
assert calls == []
def test_unproven_client_blocks_without_exception_text(monkeypatch, capsys):
module = load_script()
monkeypatch.setenv("FB_USER", "user")
monkeypatch.setenv("FB_PASSWORD", "password")
monkeypatch.setattr(module, "_client_metadata", lambda path: {
"path": str(path), "exists": True, "version": None,
})
class Driver:
def connect(self, **kwargs):
raise RuntimeError("secret server path and password")
monkeypatch.setattr(module, "_FdbDriver", lambda fdb: Driver())
assert module.main([]) == 2
output = json.loads(capsys.readouterr().out)
assert output["blocker"] == "compatibility_unproven"
assert "secret" not in json.dumps(output)
def test_firebird_authentication_error_is_classified_without_exposing_details():
module = load_script()
error = type("DatabaseError", (Exception,), {})(
"Your user name and password are not defined. SQLCODE: -902"
)
assert module._blocker_category(error) == "credentials_rejected"
+257
View File
@@ -0,0 +1,257 @@
import importlib.util
import sys
from pathlib import Path
import pytest
SCRIPT = Path(__file__).parents[1] / "scripts" / "firebird_write_executor.py"
AUTHORIZED = r"D:\PROJETOS\GTISISTEMAS\Projeto\TESTESOFICIAIS.FDB"
def load_module():
spec = importlib.util.spec_from_file_location("firebird_write_executor", SCRIPT)
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
executor = load_module()
class FakeCursor:
def __init__(self, connection):
self.connection = connection
self.rowcount = 0
self._last = None
def execute(self, sql, params=()):
self._last = (sql, params)
upper = sql.strip().upper()
if upper.startswith("SELECT F.RDB$FIELD_TYPE"):
table, field = params
info = self.connection.schema.get(table.upper(), {}).get(field.upper())
self._result = [info] if info else []
elif upper.startswith("SELECT") and "WHERE CODIGO = ?" in upper and "," in upper.split("SELECT", 1)[1].split("FROM")[0]:
code = params[0]
row = self.connection.rows.get(code)
self._result = [(code, row["IPI"])] if row else []
elif upper.startswith("SELECT IPI FROM"):
code = params[0]
row = self.connection.rows.get(code)
self._result = [(row["IPI"],)] if row else []
elif upper.startswith("UPDATE"):
matched = False
if " = NULL WHERE" in upper:
code, expected = params
row = self.connection.rows.get(code)
matched = row is not None and row["IPI"] == expected
if matched:
self.connection.pending[code] = None
elif sql.count("?") == 2:
new_value, code = params
row = self.connection.rows.get(code)
matched = row is not None and row["IPI"] is None
if matched:
self.connection.pending[code] = new_value
else:
new_value, code, expected = params
row = self.connection.rows.get(code)
matched = row is not None and row["IPI"] == expected
if matched:
self.connection.pending[code] = new_value
self.rowcount = 1 if matched else 0
self._result = []
else:
raise AssertionError(f"unexpected SQL in fake cursor: {sql}")
def fetchone(self):
return self._result[0] if self._result else None
def close(self):
pass
class FakeConnection:
def __init__(self, rows, schema):
self.rows = rows
self.schema = schema
self.pending = {}
self.committed = False
self.rolled_back = False
def cursor(self):
return FakeCursor(self)
def commit(self):
self.committed = True
for code, value in self.pending.items():
self.rows[code]["IPI"] = value
self.pending = {}
def rollback(self):
self.rolled_back = True
self.pending = {}
def _connection(ipi=None, field_length=8):
return FakeConnection(
rows={"000004": {"CODIGO": "000004", "IPI": ipi}},
schema={"C000025": {"IPI": (37, field_length)}},
)
def _proposal(**overrides):
base = {
"product_code": "000004",
"field": "IPI",
"current_value": None,
"proposed_value": "NT",
"status": "CHANGE_REQUIRED",
"source_dataset": "TIPI_IPI",
"source_version": "abc123",
"write_eligible": True,
}
base.update(overrides)
return base
def test_rejects_wrong_database_path():
with pytest.raises(executor.WriteRefused):
executor.assert_authorized_write_path(r"D:\somewhere\else.FDB")
def test_accepts_exact_authorized_path():
assert executor.assert_authorized_write_path(AUTHORIZED) == executor.AUTHORIZED_WRITE_DATABASE_PATH
def test_rejects_non_allowlisted_field():
with pytest.raises(executor.WriteRefused):
executor.assert_allowed_field("C000025", "CEST")
def test_rejects_non_allowlisted_table():
with pytest.raises(executor.WriteRefused):
executor.assert_allowed_field("OTHER_TABLE", "IPI")
def test_rejects_non_write_eligible_proposal():
connection = _connection()
with pytest.raises(executor.WriteRefused):
executor.execute_guarded_update(
connection,
database_path=AUTHORIZED,
proposal=_proposal(write_eligible=False),
now="2026-09-18T00:00:00Z",
)
assert connection.committed is False
def test_successful_write_and_verification():
connection = _connection(ipi=None)
record = executor.execute_guarded_update(
connection,
database_path=AUTHORIZED,
proposal=_proposal(),
now="2026-09-18T00:00:00Z",
)
assert record.before_value is None
assert record.after_value == "NT"
assert record.rowcount == 1
assert connection.committed is True
assert connection.rows["000004"]["IPI"] == "NT"
def test_drift_blocks_write():
connection = _connection(ipi="already-set")
with pytest.raises(executor.WriteRefused):
executor.execute_guarded_update(
connection,
database_path=AUTHORIZED,
proposal=_proposal(current_value=None),
now="2026-09-18T00:00:00Z",
)
assert connection.committed is False
assert connection.rows["000004"]["IPI"] == "already-set"
def test_wrong_database_path_never_touches_connection():
connection = _connection()
with pytest.raises(executor.WriteRefused):
executor.execute_guarded_update(
connection,
database_path=r"D:\PRODUCAO\ERP.FDB",
proposal=_proposal(),
now="2026-09-18T00:00:00Z",
)
assert connection.committed is False
assert connection.rows["000004"]["IPI"] is None
def test_restore_round_trip():
connection = _connection(ipi="NT")
record = executor.execute_guarded_restore(
connection,
database_path=AUTHORIZED,
table="C000025",
field="IPI",
key_field="CODIGO",
product_code="000004",
written_value="NT",
restore_value=None,
now="2026-09-18T00:00:00Z",
)
assert record.after_value is None
assert connection.rows["000004"]["IPI"] is None
def test_rejects_write_that_would_truncate_varchar_field():
# Mirrors the live C000025.IPI column: VARCHAR(1), while TIPI proposes "NT".
connection = _connection(ipi=None, field_length=1)
with pytest.raises(executor.WriteRefused):
executor.execute_guarded_update(
connection,
database_path=AUTHORIZED,
proposal=_proposal(proposed_value="NT"),
now="2026-09-18T00:00:00Z",
)
assert connection.committed is False
assert connection.rows["000004"]["IPI"] is None
def test_explicit_authorized_path_overrides_dev_default():
client_path = r"D:\ClienteX\ERP\PRODUCAO.FDB"
assert executor.assert_authorized_write_path(client_path, authorized_path=client_path) == Path(client_path)
def test_dev_path_is_refused_when_pilot_authorized_path_configured():
client_path = r"D:\ClienteX\ERP\PRODUCAO.FDB"
with pytest.raises(executor.WriteRefused):
executor.assert_authorized_write_path(AUTHORIZED, authorized_path=client_path)
def test_unconfigured_pilot_never_falls_back_to_authorizing_a_real_client_db():
# No explicit authorized_path passed (as if an installer forgot to set
# AUTHORIZED_WRITE_DATABASE_PATH) => falls back to the dev-only constant,
# so any real client path is still refused, not silently allowed.
with pytest.raises(executor.WriteRefused):
executor.assert_authorized_write_path(r"D:\ClienteX\ERP\PRODUCAO.FDB")
def test_schema_fingerprint_guard_requires_a_match():
with pytest.raises(executor.WriteRefused):
executor.assert_schema_fingerprint_authorized("abc", "def")
with pytest.raises(executor.WriteRefused):
executor.assert_schema_fingerprint_authorized("abc", None)
executor.assert_schema_fingerprint_authorized("abc", "abc") # does not raise
def test_null_proposed_value_refused():
connection = _connection()
with pytest.raises(executor.WriteRefused):
executor.execute_guarded_update(
connection,
database_path=AUTHORIZED,
proposal=_proposal(proposed_value=None),
now="2026-09-18T00:00:00Z",
)
+34
View File
@@ -0,0 +1,34 @@
import importlib.util
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).parents[1]
sys.path.insert(0, str(ROOT))
spec = importlib.util.spec_from_file_location("import_fiscal_context", ROOT / "scripts" / "import_fiscal_context.py")
mod = importlib.util.module_from_spec(spec); sys.modules[spec.name] = mod; spec.loader.exec_module(mod)
HEADER = "company_key,scope,scope_key,key,value,reason\n"
def test_csv_answers_are_validated_and_unanswered_rows_skipped(tmp_path, capsys):
f = tmp_path / "a.csv"
f.write_text(HEADER + "abc,GROUP,000019,product_role,RESALE,revenda\nabc,COMPANY,,entry_scope,INTRA,\nabc,GROUP,000020,product_role,,\n", encoding="utf-8")
mod.main([str(f), "--token", "t", "--dry-run"])
assert "validated 2 entries (1 unanswered rows skipped)" in capsys.readouterr().out
def test_invalid_rows_block_the_whole_import(tmp_path):
f = tmp_path / "b.csv"
f.write_text(HEADER + "abc,GROUP,000019,product_role,RESALE,\nabc,COMPANY,,rounding,Z,\n", encoding="utf-8")
with pytest.raises(SystemExit, match="invalid"):
mod.main([str(f), "--token", "t", "--dry-run"])
def test_json_format_and_questionnaire_csv_columns(tmp_path, capsys):
f = tmp_path / "c.json"
f.write_text(json.dumps({"company_key": "abc", "entries": [{"scope": "NCM", "scope_key": "22030000", "key": "cest", "value": "03.021.03"}]}), encoding="utf-8")
mod.main([str(f), "--token", "t", "--dry-run"])
assert "validated 1 entries" in capsys.readouterr().out
+58
View File
@@ -0,0 +1,58 @@
import importlib.util
import json
import sys
from pathlib import Path
ROOT = Path(__file__).parents[1]
sys.path.insert(0, str(ROOT / "agent" / "src")); sys.path.insert(0, str(ROOT / "scripts"))
spec = importlib.util.spec_from_file_location("map_product_ui_fields", ROOT / "scripts" / "map_product_ui_fields.py")
tool = importlib.util.module_from_spec(spec); sys.modules[spec.name] = tool; spec.loader.exec_module(tool)
def test_suggested_temp_is_never_the_current_value_nor_blank():
assert tool.suggest_temp("5405", ["5405", "5102"]) == "5102"
assert tool.suggest_temp(None, ["1"]) == "1"
assert tool.suggest_temp("5", ["5", "10"]) == "10" and tool.suggest_temp("5.0", ["5", "10"]) == "10"
assert tool.suggest_temp("A", ["a"]) is None
for ui, col, cands in tool.SEQUENCE:
assert cands and len(set(cands)) == len(cands)
def test_sequence_covers_all_required_ui_fields():
labels = [s[0] for s in tool.SEQUENCE]
assert len(labels) == 21 and labels[0] == "NCM" and labels[-1] == "IPPT"
def test_binding_classification_never_picks_among_several_columns():
d = lambda *f, tech=(): [{"FIELD": x, "BEFORE": 1, "AFTER": 2, "technical": False} for x in f] + [{"FIELD": x, "BEFORE": 1, "AFTER": 2, "technical": True} for x in tech] # noqa: E731
assert tool.classify(d("CFOP", tech=("ULTIMA_ALTERACAO",)), "CFOP")[0] == "DIRECT_BINDING"
assert tool.classify(d("CFOP", "CFOP_CUPOM"), "CFOP")[0] == "DERIVED/COUPLED_FIELD"
assert tool.classify(d("A", "B"), "CFOP")[0] == "AMBIGUOUS"
assert tool.classify(d(tech=("ULTIMA_ALTERACAO",)), "CFOP")[0] == "AUTOMATIC_METADATA"
assert tool.classify([], "CFOP")[0] == "AMBIGUOUS"
assert tool.classify(d("OTHER"), "CFOP")[0] == "DIRECT_BINDING" # unexpected column is reported, not hidden
def test_step_records_diff_updates_baseline_and_persists(tmp_path, monkeypatch):
monkeypatch.setattr(tool, "SESSION", tmp_path / "session.json")
session = tool.new_session("000001"); session["baseline"] = {"CFOP": "5102", "CSOSN": "102", "ULTIMA_ALTERACAO": "t0"}
after = {"CFOP": "5405", "CSOSN": "102", "ULTIMA_ALTERACAO": "t1"}
out = tool.run_step(session, "CFOP saída", "CFOP", ["5405"], lambda: after, lambda prompt: "")
rec = session["results"][0]
assert out == "tested" and rec["binding_type"] == "DIRECT_BINDING" and rec["db_fields_changed"] == ["CFOP"] and rec["technical_fields_changed"] == ["ULTIMA_ALTERACAO"]
assert session["baseline"] == after # new baseline for the next field
assert json.loads((tmp_path / "session.json").read_text(encoding="utf-8"))["results"][0]["ui_label"] == "CFOP saída"
assert session["firebird_writes"] == 0
def test_skip_and_quit(tmp_path, monkeypatch):
monkeypatch.setattr(tool, "SESSION", tmp_path / "s.json")
session = tool.new_session("1"); session["baseline"] = {}
assert tool.run_step(session, "CEST", "CEST", ["1"], lambda: {}, lambda p: "s") == "skipped" and session["results"][0]["status"] == "NOT_ON_SCREEN"
assert tool.run_step(session, "X", "X", ["1"], lambda: {}, lambda p: "q") == "quit"
def test_tool_never_opens_a_write_connection():
src = (ROOT / "scripts" / "map_product_ui_fields.py").read_text(encoding="utf-8") + (ROOT / "scripts" / "sgv_product_snapshot.py").read_text(encoding="utf-8")
for forbidden in ("UPDATE ", "INSERT ", "DELETE FROM", "commit(", "execute_guarded"):
assert forbidden not in src
+253
View File
@@ -0,0 +1,253 @@
from gti_discovery.mapping import build_schema_mapping, identify_product_candidates
def fixture_metadata():
return {
"relations": [
{"name": "PRODUCT", "approximate_count": 12},
{"name": "PRODUCT_STOCK", "approximate_count": 12},
{"name": "PRODUCT_IMAGE"},
],
"fields": [
{"relation": "PRODUCT", "name": "ID", "type": "INTEGER", "nullable": False},
{"relation": "PRODUCT", "name": "CODE", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "DESCRIPTION", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "NCM", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "IAT", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "IPPT", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "INCIDENCIA", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "CFOP_ENTRADA", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "CFOP_SAIDA", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "CST_ENTRADA", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "CST_SAIDA", "type": "VARCHAR"},
{"relation": "PRODUCT", "name": "IMAGE_DATA", "type": "BLOB", "subtype": 0},
{"relation": "PRODUCT_STOCK", "name": "PRODUCT_ID", "type": "INTEGER"},
{"relation": "PRODUCT_STOCK", "name": "QTY", "type": "DECIMAL"},
{"relation": "PRODUCT_IMAGE", "name": "PRODUCT_ID", "type": "INTEGER"},
{"relation": "PRODUCT_IMAGE", "name": "IMAGE_URL", "type": "VARCHAR"},
],
"keys": [
{"relation": "PRODUCT", "constraint_type": "PRIMARY KEY", "fields": ["ID"]},
],
"foreign_keys": [
{"child_relation": "PRODUCT_STOCK", "child_field": "PRODUCT_ID", "referenced_relation": "PRODUCT", "referenced_field": "ID"},
{"child_relation": "PRODUCT_IMAGE", "child_field": "PRODUCT_ID", "referenced_relation": "PRODUCT", "referenced_field": "ID"},
],
"indexes": [{"relation": "PRODUCT", "fields": ["CODE"], "unique": True}],
"triggers": [{"relation": "PRODUCT", "name": "BI_PRODUCT", "timing": "BEFORE", "event": "INSERT"}],
"generators": [{"name": "GEN_PRODUCT_ID"}],
"dependencies": [{"dependent": "BI_PRODUCT", "depended_on": "GEN_PRODUCT_ID"}],
}
def test_product_is_confirmed_only_when_structure_and_masked_scalar_samples_agree():
samples = {"PRODUCT": [{"ID": 1, "CODE": "P-001", "DESCRIPTION": "masked"}]}
candidates = identify_product_candidates(fixture_metadata(), samples)
product = next(item for item in candidates if item["relation"] == "PRODUCT")
assert product["classification"] == "CONFIRMED"
assert product["evidence"]["primary_key"] == ["ID"]
def test_mapping_prefers_primary_product_structure_over_auxiliary_product_table():
metadata = {
"relations": [
{"name": "PRODUTOS_AUX"},
{"name": "MATERIAL"},
],
"fields": [
{"relation": "PRODUTOS_AUX", "name": "ID", "type": "INTEGER"},
{"relation": "PRODUTOS_AUX", "name": "CODIGO", "type": "VARCHAR"},
{"relation": "MATERIAL", "name": "MATERIAL_CODIGO", "type": "VARCHAR"},
{"relation": "MATERIAL", "name": "MATERIAL_NOME", "type": "VARCHAR"},
{"relation": "MATERIAL", "name": "MATERIAL_UNIDADE", "type": "VARCHAR"},
{"relation": "MATERIAL", "name": "MATERIAL_ESTOQUE", "type": "DECIMAL"},
],
"keys": [
{"relation": "PRODUTOS_AUX", "constraint_type": "PRIMARY KEY", "fields": ["ID"]},
{"relation": "MATERIAL", "constraint_type": "PRIMARY KEY", "fields": ["MATERIAL_CODIGO"]},
],
}
result = build_schema_mapping(metadata, {
"PRODUTOS_AUX": [{"ID": 1, "CODIGO": "masked"}],
"MATERIAL": [{"MATERIAL_CODIGO": "masked", "MATERIAL_NOME": "masked", "MATERIAL_UNIDADE": "masked", "MATERIAL_ESTOQUE": 1}],
})
assert result["product"]["relation"] == "MATERIAL"
assert result["product"]["evidence"]["primary_key"] == ["MATERIAL_CODIGO"]
def test_inferred_product_preserves_structural_evidence():
metadata = {
"relations": [{"name": "MATERIAL"}],
"fields": [
{"relation": "MATERIAL", "name": "MATERIAL_CODIGO", "type": "VARCHAR"},
{"relation": "MATERIAL", "name": "MATERIAL_NOME", "type": "VARCHAR"},
],
"keys": [{"relation": "MATERIAL", "constraint_type": "PRIMARY KEY", "fields": ["MATERIAL_CODIGO"]}],
}
result = build_schema_mapping(metadata, {})
assert result["product"]["classification"] == "INFERRED"
assert result["product"]["evidence"]["primary_key"] == ["MATERIAL_CODIGO"]
def test_mapping_prefers_sampled_product_fields_over_empty_structural_table():
metadata = {
"relations": [{"name": "C000119"}, {"name": "CL00007"}],
"fields": [
{"relation": "C000119", "name": "CODIGO", "type": "INTEGER"},
{"relation": "C000119", "name": "NCM", "type": "VARCHAR"},
{"relation": "C000119", "name": "ESTOQUE", "type": "DECIMAL"},
{"relation": "CL00007", "name": "CODIGO", "type": "INTEGER"},
{"relation": "CL00007", "name": "PRODUTO", "type": "VARCHAR"},
{"relation": "CL00007", "name": "UNIDADE", "type": "VARCHAR"},
{"relation": "CL00007", "name": "MARCA", "type": "VARCHAR"},
{"relation": "CL00007", "name": "ESTOQUE", "type": "DECIMAL"},
],
"keys": [],
}
result = build_schema_mapping(metadata, {
"C000119": [],
"CL00007": [{"CODIGO": 276, "PRODUTO": "masked", "UNIDADE": "UN", "MARCA": "masked", "ESTOQUE": 1}],
})
assert result["product"]["relation"] == "CL00007"
def test_name_only_product_candidate_is_inferred_and_absence_is_unknown():
inferred = identify_product_candidates({"relations": [{"name": "PRODUCT"}], "fields": []}, {})
assert inferred[0]["classification"] == "INFERRED"
unknown = identify_product_candidates({"relations": [{"name": "LEDGER"}], "fields": []}, {})
assert unknown[0]["classification"] == "UNKNOWN"
def test_generic_primary_key_and_sample_do_not_confirm_unrelated_table():
metadata = {
"relations": [{"name": "PRODUCT"}],
"fields": [
{"relation": "PRODUCT", "name": "ID", "type": "INTEGER"},
{"relation": "PRODUCT", "name": "VALUE", "type": "VARCHAR"},
],
"keys": [{"relation": "PRODUCT", "constraint_type": "PRIMARY KEY", "fields": ["ID"]}],
}
candidate = identify_product_candidates(
metadata, {"PRODUCT": [{"ID": 1, "VALUE": "unrelated"}]}
)[0]
assert candidate["classification"] == "INFERRED"
assert candidate["evidence"]["semantic_fields"] == []
def test_mapping_covers_fiscal_media_stock_incremental_and_redacts_sensitive_samples():
samples = {"PRODUCT": [{"ID": 1, "NCM": "masked", "IMAGE_DATA": b"raw blob", "PASSWORD": "secret"}]}
result = build_schema_mapping(fixture_metadata(), samples)
assert result["product"]["classification"] == "CONFIRMED"
assert result["fiscal"]["fields"]["NCM"]["classification"] in {"CONFIRMED", "INFERRED"}
assert result["stock"]["relations"]["PRODUCT_STOCK"]["classification"] == "CONFIRMED"
assert result["media"]["fields"]["IMAGE_DATA"]["value"] == "<redacted>"
assert result["incremental"]["classification"] == "INFERRED"
assert result["triggers"][0]["source"] == "sanitized metadata only"
assert "raw blob" not in str(result)
assert "secret" not in str(result)
def test_mapping_recognizes_explicit_fiscal_concepts_and_operation_directions():
result = build_schema_mapping(
fixture_metadata(),
{"PRODUCT": [{
"ID": 1,
"IAT": "masked",
"IPPT": "masked",
"INCIDENCIA": "masked",
"CFOP_ENTRADA": "masked",
"CFOP_SAIDA": "masked",
"CST_ENTRADA": "masked",
"CST_SAIDA": "masked",
}]},
)
assert set(result["fiscal"]["fields"]) >= {
"IAT", "IPPT", "INCIDENCIA",
"CFOP_ENTRADA", "CFOP_SAIDA", "CST_ENTRADA", "CST_SAIDA",
}
def test_missing_mapping_evidence_is_unknown():
result = build_schema_mapping({"relations": [], "fields": []}, {})
assert result["product"]["classification"] == "UNKNOWN"
assert result["fiscal"]["classification"] == "UNKNOWN"
assert result["media"]["classification"] == "UNKNOWN"
assert result["incremental"]["classification"] == "UNKNOWN"
def collector_shaped_metadata():
return {
"relations": [
{"RDB$RELATION_NAME": "PRODUCT"},
{"RDB$RELATION_NAME": "PRODUCT_STOCK"},
{"RDB$RELATION_NAME": "PRODUCT_IMAGE"},
{"RDB$RELATION_NAME": "PRODUCT_FISCAL"},
],
"fields": [
{"RELATION_NAME": "PRODUCT", "FIELD_NAME": "ID", "TYPE_NAME": "INTEGER"},
{"RELATION_NAME": "PRODUCT", "FIELD_NAME": "DESCRIPTION", "TYPE_NAME": "VARCHAR"},
{"RELATION_NAME": "PRODUCT_STOCK", "FIELD_NAME": "PRODUCT_ID", "TYPE_NAME": "INTEGER"},
{"RELATION_NAME": "PRODUCT_STOCK", "FIELD_NAME": "QUANTITY", "TYPE_NAME": "DECIMAL"},
{"RELATION_NAME": "PRODUCT_IMAGE", "FIELD_NAME": "PRODUCT_ID", "TYPE_NAME": "INTEGER"},
{"RELATION_NAME": "PRODUCT_IMAGE", "FIELD_NAME": "IMAGE_DATA", "TYPE_NAME": "BLOB", "FIELD_SUB_TYPE": 0},
{"RELATION_NAME": "PRODUCT_IMAGE", "FIELD_NAME": "IMAGE_URL", "TYPE_NAME": "VARCHAR"},
{"RELATION_NAME": "PRODUCT_FISCAL", "FIELD_NAME": "PRODUCT_ID", "TYPE_NAME": "INTEGER"},
{"RELATION_NAME": "PRODUCT_FISCAL", "FIELD_NAME": "NCM", "TYPE_NAME": "VARCHAR"},
],
"keys": [{"RELATION_NAME": "PRODUCT", "CONSTRAINT_TYPE": "PRIMARY KEY", "FIELD_NAME": "ID"}],
"foreign_keys": [
{"CHILD_RELATION_NAME": "PRODUCT_STOCK", "CHILD_FIELD_NAME": "PRODUCT_ID", "REFERENCED_RELATION_NAME": "PRODUCT", "REFERENCED_FIELD_NAME": "ID"},
{"CHILD_RELATION_NAME": "PRODUCT_IMAGE", "CHILD_FIELD_NAME": "PRODUCT_ID", "REFERENCED_RELATION_NAME": "PRODUCT", "REFERENCED_FIELD_NAME": "ID"},
{"CHILD_RELATION_NAME": "PRODUCT_FISCAL", "CHILD_FIELD_NAME": "PRODUCT_ID", "REFERENCED_RELATION_NAME": "PRODUCT", "REFERENCED_FIELD_NAME": "ID"},
],
"triggers": [{"TRIGGER_NAME": "BI_PRODUCT", "RELATION_NAME": "PRODUCT", "TRIGGER_TYPE": 1}],
"dependencies": [{"DEPENDENT_NAME": "BI_PRODUCT", "DEPENDED_ON_NAME": "GEN_PRODUCT_ID"}],
"generators": [{"GENERATOR_NAME": "GEN_PRODUCT_ID"}],
}
def test_collector_shapes_map_separate_relations_and_correlate_dependencies():
result = build_schema_mapping(
collector_shaped_metadata(),
{"PRODUCT": [{"ID": 1, "DESCRIPTION": "customer item"}], "PRODUCT_IMAGE": [{"IMAGE_DATA": b"private"}]},
)
assert "PRODUCT_STOCK" in result["stock"]["relations"]
assert "PRODUCT_IMAGE" in result["media"]["relations"]
assert result["media"]["relations"]["PRODUCT_IMAGE"]["fields"]["IMAGE_DATA"]["subtype"] == 0
assert "PRODUCT_FISCAL" in result["fiscal"]["relations"]
assert result["triggers"][0]["trigger_type"] == 1
assert result["triggers"][0]["dependencies"] == ["GEN_PRODUCT_ID"]
def test_null_trigger_type_is_reported_as_unknown():
result = build_schema_mapping(
{"relations": [{"name": "PRODUCT"}], "fields": [], "triggers": [{"relation": "PRODUCT", "name": "BI_PRODUCT", "trigger_type": None}]},
{},
)
assert result["triggers"][0]["trigger_type"] == "UNKNOWN"
def test_samples_expose_only_allowlisted_structural_evidence():
result = build_schema_mapping(
fixture_metadata(),
{"PRODUCT": [{"ID": 1, "CODE": "customer-commercial-code", "DESCRIPTION": "customer name", "NCM": "12345678"}]},
)
serialized = str(result)
assert "customer-commercial-code" not in serialized
assert "customer name" not in serialized
assert "12345678" not in serialized
assert result["product"]["evidence"]["sample_fields"] == ["CODE", "DESCRIPTION", "ID", "NCM"]
+173
View File
@@ -0,0 +1,173 @@
import importlib.metadata
import importlib.util
from pathlib import Path
import pytest
from gti_discovery import provenance
def _patch_environment(monkeypatch, tmp_path, *, origin=None, version="2.0.4"):
project_root = tmp_path / "project"
project_root.mkdir()
site_packages = tmp_path / "approved-site-packages"
site_packages.mkdir()
module_file = origin or site_packages / "fdb" / "__init__.py"
module_file.parent.mkdir(parents=True, exist_ok=True)
module_file.write_text("# fixture", encoding="ascii")
distribution = {
"Name": "fdb",
"Version": version,
"Location": str(site_packages),
}
monkeypatch.setattr(
importlib.util,
"find_spec",
lambda name: type("Spec", (), {
"origin": str(module_file),
"submodule_search_locations": [str(module_file.parent)],
})() if name == "fdb" else None,
)
monkeypatch.setattr(importlib.metadata, "version", lambda name: version)
monkeypatch.setattr(importlib.metadata, "distribution", lambda name: type(
"Distribution", (), {
"metadata": distribution,
"locate_file": lambda self, value: site_packages / value,
}
)())
monkeypatch.setattr(provenance, "PROJECT_ROOT", project_root)
monkeypatch.setattr(provenance, "APPROVED_SITE_PACKAGES_ROOT", site_packages)
monkeypatch.setattr(provenance.sys, "path", [str(site_packages)])
monkeypatch.setenv("PYTHONPATH", str(site_packages))
return site_packages, module_file
def test_inspection_correlates_module_and_distribution_without_importing_fdb(monkeypatch, tmp_path):
site_packages, module_file = _patch_environment(monkeypatch, tmp_path)
evidence = provenance.inspect_fdb_provenance()
assert evidence["module_file"] == str(module_file.resolve())
assert evidence["module_package"] == "fdb"
assert evidence["module_version"] == "2.0.4"
assert evidence["distribution_name"] == "fdb"
assert evidence["distribution_version"] == "2.0.4"
assert evidence["distribution_location"] == str(site_packages.resolve())
assert evidence["module_under_distribution"] is True
assert evidence["shadowing"] == []
assert evidence["status"] == "VALID"
def test_validation_accepts_matching_installed_provenance(monkeypatch, tmp_path):
_patch_environment(monkeypatch, tmp_path)
evidence = provenance.validate_fdb_provenance()
assert evidence["status"] == "VALID"
assert evidence["validated_version"] == "2.0.4"
@pytest.mark.parametrize("shadow", ["fdb.py", "fdb"])
def test_validation_rejects_project_local_shadowing(monkeypatch, tmp_path, shadow):
_patch_environment(monkeypatch, tmp_path)
local = provenance.PROJECT_ROOT / shadow
if shadow == "fdb":
local.mkdir()
(local / "__init__.py").write_text("# shadow", encoding="ascii")
else:
local.write_text("# shadow", encoding="ascii")
with pytest.raises(provenance.ProvenanceError, match="shadow"):
provenance.validate_fdb_provenance()
def test_validation_rejects_project_pth_candidate(monkeypatch, tmp_path):
_patch_environment(monkeypatch, tmp_path)
(provenance.PROJECT_ROOT / "unexpected.pth").write_text("evil", encoding="ascii")
with pytest.raises(provenance.ProvenanceError, match="shadow"):
provenance.validate_fdb_provenance()
def test_validation_blocks_mismatched_distribution(monkeypatch, tmp_path):
_patch_environment(monkeypatch, tmp_path, version="2.0.3")
with pytest.raises(provenance.ProvenanceError, match="version"):
provenance.validate_fdb_provenance()
def test_validation_blocks_ambiguous_origin(monkeypatch, tmp_path):
site_packages, module_file = _patch_environment(monkeypatch, tmp_path)
duplicate = site_packages / "fdb.py"
duplicate.write_text("# duplicate", encoding="ascii")
with pytest.raises(provenance.ProvenanceError, match="ambiguous"):
provenance.validate_fdb_provenance()
def test_validation_rejects_distribution_outside_approved_environment(monkeypatch, tmp_path):
_patch_environment(monkeypatch, tmp_path)
external = tmp_path / "unapproved-site-packages"
external.mkdir()
module_file = external / "fdb" / "__init__.py"
module_file.parent.mkdir()
module_file.write_text("# external", encoding="ascii")
monkeypatch.setattr(importlib.util, "find_spec", lambda name: type(
"Spec", (), {"origin": str(module_file), "name": "fdb"}
)())
monkeypatch.setattr(importlib.metadata, "distribution", lambda name: type(
"Distribution", (), {
"metadata": {"Name": "fdb", "Version": "2.0.4"},
"locate_file": lambda self, value: external / value,
}
)())
with pytest.raises(provenance.ProvenanceError, match="approved"):
provenance.validate_fdb_provenance()
def test_validation_rejects_shadowing_under_project_site_packages(monkeypatch, tmp_path):
_patch_environment(monkeypatch, tmp_path)
local = provenance.PROJECT_ROOT / "site-packages" / "fdb.py"
local.parent.mkdir(parents=True)
local.write_text("# shadow", encoding="ascii")
with pytest.raises(provenance.ProvenanceError, match="shadow"):
provenance.validate_fdb_provenance()
def test_post_import_validation_checks_actual_module_attributes(monkeypatch, tmp_path):
_patch_environment(monkeypatch, tmp_path)
imported = type("Fdb", (), {
"__file__": str(tmp_path / "wrong" / "fdb.py"),
"__package__": "wrong",
"__version__": "2.0.4",
})()
with pytest.raises(provenance.ProvenanceError, match="imported"):
provenance.validate_imported_fdb_provenance(imported)
def test_post_import_validation_accepts_matching_module_and_version(monkeypatch, tmp_path):
_, module_file = _patch_environment(monkeypatch, tmp_path)
imported = type("Fdb", (), {
"__file__": str(module_file),
"__package__": "fdb",
"__version__": "2.0.4",
})()
evidence = provenance.validate_imported_fdb_provenance(imported)
assert evidence["status"] == "VALID"
assert evidence["imported_module_file"] == str(module_file.resolve())
def test_post_import_validation_rejects_missing_version(monkeypatch, tmp_path):
_, module_file = _patch_environment(monkeypatch, tmp_path)
imported = type("Fdb", (), {
"__file__": str(module_file),
"__package__": "fdb",
})()
with pytest.raises(provenance.ProvenanceError, match="version"):
provenance.validate_imported_fdb_provenance(imported)
+110
View File
@@ -0,0 +1,110 @@
import pytest
from gti_discovery.read_only import validate_read_only
def test_accepts_and_normalizes_select_query():
assert validate_read_only(" SELECT * FROM RDB$DATABASE ; ") == (
"SELECT * FROM RDB$DATABASE"
)
def test_accepts_with_query():
sql = "WITH x AS (SELECT 1 AS n FROM RDB$DATABASE) SELECT n FROM x"
assert validate_read_only(sql) == sql
@pytest.mark.parametrize(
"sql",
[
"INSERT INTO X VALUES (1)",
"UPDATE X SET A=1",
"DELETE FROM X",
"CREATE TABLE X(A INT)",
"ALTER TABLE X ADD A INT",
"DROP TABLE X",
"TRUNCATE TABLE X",
"EXECUTE PROCEDURE P",
"WITH x AS (SELECT 1 FROM RDB$DATABASE) INSERT INTO X VALUES (1)",
],
)
def test_rejects_write_or_side_effect_sql(sql):
with pytest.raises(ValueError):
validate_read_only(sql)
@pytest.mark.parametrize(
"sql",
[
"SELECT 1 FROM RDB$DATABASE; SELECT 2 FROM RDB$DATABASE",
"SELECT 1 FROM RDB$DATABASE -- comment",
"SELECT 1 /* comment */ FROM RDB$DATABASE",
"",
"VACUUM TABLE X",
],
)
def test_rejects_multiple_statements_comments_and_non_queries(sql):
with pytest.raises(ValueError):
validate_read_only(sql)
@pytest.mark.parametrize(
"sql",
[
"SELECT * FROM SIDE_EFFECTING_PROC(1)",
"SELECT GEN_ID(MY_GENERATOR, 1) FROM RDB$DATABASE",
"SELECT NEXT VALUE FOR MY_GENERATOR FROM RDB$DATABASE",
"SELECT * FROM X WITH LOCK",
"SELECT * FROM X FOR UPDATE",
"SELECT * FROM PKG . SELECTABLE_PROC (1)",
],
)
def test_rejects_known_side_effecting_select_constructs(sql):
with pytest.raises(ValueError):
validate_read_only(sql)
@pytest.mark.parametrize(
"sql",
[
"SELECT IIF(RDB$SYSTEM_FLAG = 1, 'SYSTEM', 'USER') FROM RDB$RELATIONS",
"SELECT EXTRACT(YEAR FROM CURRENT_DATE) FROM RDB$DATABASE",
"SELECT COALESCE(RDB$RELATION_NAME, '') FROM RDB$RELATIONS",
],
)
def test_accepts_legitimate_firebird_read_only_expressions(sql):
assert validate_read_only(sql) == sql
@pytest.mark.parametrize(
"sql",
[
"SELECT EXECUTE(1) FROM RDB$DATABASE",
"SELECT GEN_ID(MY_GENERATOR, 1) FROM RDB$DATABASE",
"SELECT * FROM SELECTABLE_PROC(1)",
"SELECT * FROM PKG . SELECTABLE_PROC (1)",
"SELECT * FROM PKG\t.\tSELECTABLE_PROC\t(1)",
],
)
def test_rejects_explicit_unsafe_callable_forms(sql):
with pytest.raises(ValueError):
validate_read_only(sql)
@pytest.mark.parametrize(
"sql",
[
"SELECT SIDE_EFFECTING_UDF(ID) FROM X",
"SELECT PKG . SIDE_EFFECTING_UDF(ID) FROM X",
],
)
def test_rejects_unknown_callable_in_select_projection(sql):
with pytest.raises(ValueError, match="callable"):
validate_read_only(sql)
def test_normalization_does_not_change_quoted_literal_contents():
sql = " SELECT ' keep spaces ; -- text ' FROM RDB$DATABASE ; "
assert validate_read_only(sql) == (
"SELECT ' keep spaces ; -- text ' FROM RDB$DATABASE"
)
+199
View File
@@ -0,0 +1,199 @@
import pytest
import gti_discovery
from gti_discovery.catalog import FirebirdCatalog
from gti_discovery.collector import _approved_fixture_reader
from gti_discovery.firebird_queries import (
_ADAPTER_REGISTRATION_TOKEN,
_APPROVED_ADAPTER_TYPES,
_CATALOG_QUERIES,
_INTERNAL_QUERIES,
_execute_catalog,
_register_approved_adapter,
)
from gti_discovery.collector import _APPROVED_FIXTURE_TYPES
class RecordingAdapter:
read_only = True
def __init__(self):
self.calls = []
def execute(self, sql, params=()):
self.calls.append((sql, params))
return []
_register_approved_adapter(RecordingAdapter, _ADAPTER_REGISTRATION_TOKEN)
def make_structured_connection():
return FirebirdCatalog(RecordingAdapter())
def test_public_connection_has_no_sql_execute_method():
connection = make_structured_connection()
for name in ("execute", "query", "raw_execute", "execute_statement", "run_sql", "_run"):
assert not hasattr(connection, name)
assert name not in gti_discovery.__all__
def test_package_exports_no_raw_sql_or_parser_authority():
forbidden = {
"execute", "query", "raw_execute", "execute_statement", "run_sql",
"validate_read_only",
}
assert forbidden.isdisjoint(vars(gti_discovery))
assert forbidden.isdisjoint(gti_discovery.__all__)
def test_private_sql_runner_cannot_bypass_operation_boundary():
connection = make_structured_connection()
assert not hasattr(connection, "_run")
assert not hasattr(connection, "_run_sql")
def test_unknown_operation_key_never_reaches_driver():
adapter = RecordingAdapter()
with pytest.raises(ValueError):
_execute_catalog(adapter, "SELECT * FROM X")
assert adapter.calls == []
@pytest.mark.parametrize(
"forbidden_operation",
["execute_procedure", "select_procedure", "execute_selectable_procedure", "invoke_udf"],
)
def test_procedure_and_udf_operations_never_reach_driver(forbidden_operation):
adapter = RecordingAdapter()
with pytest.raises(ValueError, match="unknown catalog operation"):
_execute_catalog(adapter, forbidden_operation)
assert adapter.calls == []
def test_procedure_api_is_metadata_only():
adapter = RecordingAdapter()
catalog = FirebirdCatalog(adapter)
assert catalog.get_procedures() == []
assert catalog.get_procedure_parameters() == []
assert not any("EXECUTE PROCEDURE" in sql.upper() for sql, _ in adapter.calls)
assert not any("FROM " + name in sql.upper() for name in ("SP_RECALC", "SOME_UDF") for sql, _ in adapter.calls)
@pytest.mark.parametrize("operation_key", [None, 1, [], {}, bytearray(b"database"), str("database") + ""])
def test_operation_key_must_be_an_exact_approved_string(operation_key):
adapter = RecordingAdapter()
if type(operation_key) is str:
pytest.skip("control case is tested by the static operation tests")
with pytest.raises(ValueError, match="operation"):
_execute_catalog(adapter, operation_key)
assert adapter.calls == []
def test_str_subclass_is_not_an_operation_key():
class Lookalike(str):
pass
adapter = RecordingAdapter()
with pytest.raises(ValueError, match="operation"):
_execute_catalog(adapter, Lookalike("database"))
assert adapter.calls == []
def test_arbitrary_adapter_is_rejected_before_driver_invocation():
class ArbitraryAdapter:
def execute(self, sql, params=()):
raise AssertionError("driver must not be invoked")
with pytest.raises(TypeError, match="approved implementation"):
_execute_catalog(ArbitraryAdapter(), "database")
def test_catalog_rejects_unregistered_read_only_adapter_before_driver_invocation():
class LookalikeAdapter:
read_only = True
def execute(self, sql, params=()):
raise AssertionError("driver must not be invoked")
with pytest.raises(TypeError, match="approved implementation"):
FirebirdCatalog(LookalikeAdapter())
def test_catalog_close_cannot_act_on_unregistered_adapter():
class LookalikeAdapter:
read_only = True
closed = False
def execute(self, sql, params=()):
raise AssertionError("driver must not be invoked")
def close(self):
self.closed = True
adapter = LookalikeAdapter()
with pytest.raises(TypeError, match="approved implementation"):
FirebirdCatalog(adapter)
assert adapter.closed is False
def test_fixture_reader_registration_requires_private_capability():
class UntrustedReader:
pass
with pytest.raises(TypeError):
_approved_fixture_reader(UntrustedReader)
def test_catalog_queries_are_immutable_and_operation_keyed():
with pytest.raises(TypeError):
_CATALOG_QUERIES["new"] = "SELECT 1"
assert set(_CATALOG_QUERIES) == {
"database", "relations", "fields", "keys", "indexes", "foreign_keys",
"triggers", "procedures", "procedure_parameters", "packages", "generators",
"dependencies", "monitoring",
}
def test_private_approved_type_registries_are_immutable_snapshots():
assert isinstance(_APPROVED_ADAPTER_TYPES, type(_CATALOG_QUERIES))
assert isinstance(_APPROVED_FIXTURE_TYPES, type(_CATALOG_QUERIES))
with pytest.raises(TypeError):
_APPROVED_ADAPTER_TYPES[RecordingAdapter] = True
with pytest.raises(TypeError):
_APPROVED_FIXTURE_TYPES[RecordingAdapter] = True
def test_registration_mutators_are_not_publicly_exported():
for name in ("_register_approved_adapter", "_approved_fixture_reader"):
assert name not in gti_discovery.__all__
assert not hasattr(gti_discovery, name)
def test_threat_model_documents_accidental_boundary_scope():
from pathlib import Path
state = Path("docs/AI_PROJECT_STATE.md").read_text(encoding="utf-8")
report = Path("docs/MISSION_00_REPORT.md").read_text(encoding="utf-8")
required = (
"implementation mistakes", "accidental unauthorized operations",
"arbitrary SQL", "wrong database", "writes", "procedures/UDFs",
"injection", "dependency shadowing",
)
for phrase in required:
assert phrase in state
assert phrase in report
assert "arbitrary Python" in state
assert "same process" in report
def test_legitimate_firebird_syntax_reaches_driver_without_old_parser(monkeypatch):
adapter = RecordingAdapter()
query = _INTERNAL_QUERIES["expression_probe"]
monkeypatch.setattr("gti_discovery.read_only.validate_read_only", lambda _: (_ for _ in ()).throw(AssertionError()))
_execute_catalog(adapter, "expression_probe")
assert adapter.calls == [(query, ())]
+79
View File
@@ -0,0 +1,79 @@
import importlib.util
import json
import re
import sys
import zipfile
from pathlib import Path
import pytest
ROOT = Path(__file__).parents[1]
sys.path.insert(0, str(ROOT))
def test_versions_are_coherent_across_hub_agent_and_packages():
from central.app.version import GTI_FISCAL_VERSION
v = (ROOT / "release" / "VERSION").read_text(encoding="utf-8").strip()
assert v == GTI_FISCAL_VERSION == "1.0.0-rc1"
pep = v.replace("-rc", "rc")
for pp in ("central/pyproject.toml", "agent/pyproject.toml"):
assert f'version = "{pep}"' in (ROOT / pp).read_text(encoding="utf-8")
assert '"AGENT_VERSION", "1.0.0-rc1"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8")
manifest = ROOT / "release" / "RELEASE_MANIFEST.json"
if manifest.exists():
m = json.loads(manifest.read_text(encoding="utf-8"))
assert m["GTI_FISCAL_VERSION"] == v and m["migrations"]["alembic_head_in_repo"] == m["migrations"]["applied_revision"]
def test_production_compose_never_exposes_postgres_or_dev_token():
import yaml
raw = (ROOT / "deploy" / "hub" / "docker-compose.prod.yml").read_text(encoding="utf-8")
doc = yaml.safe_load(raw)
pg, hub = doc["services"]["postgres"], doc["services"]["fiscal-hub"]
assert "ports" not in pg and pg["networks"] == ["internal"] and doc["networks"]["internal"]["internal"] is True
assert any(v.startswith("postgres_data:") for v in pg["volumes"]) and pg["restart"] == "unless-stopped" and "healthcheck" in pg
assert "AGENT_DEV_TOKEN" not in hub["environment"] and hub["environment"]["APP_ENV"] == "production"
assert hub["restart"] == "unless-stopped" and "healthcheck" in hub and hub["depends_on"]["postgres"]["condition"] == "service_healthy"
assert "POSTGRES_PASSWORD:?" in raw and hub["ports"][0].startswith("${HUB_BIND:-127.0.0.1}")
assert set(doc["services"]) == {"postgres", "fiscal-hub"}
images = " ".join(str(v.get("image", "")) for v in doc["services"].values()).lower()
for banned in ("redis", "rabbitmq", "kubernetes"):
assert banned not in images
def test_templates_and_env_examples_contain_no_secrets():
files = list((ROOT / "deploy").rglob("*.example")) + [ROOT / "deploy" / "agent-windows" / "agent.env.template", ROOT / "deploy" / "hub" / ".env.example", ROOT / ".env.example"]
for f in files:
text = f.read_text(encoding="utf-8")
for m in re.finditer(r"^(AGENT_TOKEN|FB_PASSWORD|POSTGRES_PASSWORD|FB_USER)=(.*)$", text, re.M):
assert m.group(2).strip() == "", f"{f.name}: {m.group(1)} must be empty in a template"
pilot = (ROOT / "deploy" / "agent-windows" / "pilot-construcao.env.example").read_text(encoding="utf-8")
assert "COMPANY_CODE=000001" in pilot and "COMPANY_CRT_SOURCE=AUTO" in pilot and "FISCAL_MODE=DRY_RUN" in pilot and "FIREBIRD_WRITES=0" in pilot
for name in ("mmmodas", "rincao", "uniao"):
prof = (ROOT / "deploy" / "agent-windows" / "profiles" / f"{name}.env.example").read_text(encoding="utf-8")
assert "COMPANY_CRT_SOURCE=EMISSOR" in prof and "FISCAL_MODE=DRY_RUN" in prof and "FIREBIRD_WRITES=0" in prof
def test_agent_package_is_self_contained_and_secret_free(tmp_path):
spec = importlib.util.spec_from_file_location("build_agent_package", ROOT / "scripts" / "build_agent_package.py")
mod = importlib.util.module_from_spec(spec); sys.modules[spec.name] = mod; spec.loader.exec_module(mod)
mod.ROOT = ROOT
mod.main()
z = zipfile.ZipFile(ROOT / "dist" / f"{mod.NAME}.zip")
names = z.namelist()
assert len(names) == len(set(names))
for required in ("gti-agent.ps1", "gti-agent.cmd", "requirements-agent.txt", "pilot-construcao.env.example", "agent/scripts/preflight.py", "agent/scripts/run_agent_service.py",
"agent/service/gti_agent_windows_service.py", "scripts/firebird_write_executor.py", "central/app/fiscal_decision.py", "MANIFEST.json", "VERSION"):
assert f"{mod.NAME}/{required}" in names, required
assert not [n for n in names if n.endswith((".env", ".sqlite", ".log", ".FDB"))]
ps = z.read(f"{mod.NAME}/gti-agent.ps1").decode("utf-8")
for cmd in ("install", "configure", "preflight", "start", "stop", "restart", "status", "uninstall"):
assert f"'{cmd}'" in ps
def test_secret_scan_rejects_a_token_in_a_template():
spec = importlib.util.spec_from_file_location("bap2", ROOT / "scripts" / "build_agent_package.py")
mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
with pytest.raises(SystemExit):
mod.scan("x.env.example", b"AGENT_TOKEN=abcdefghijklmnop123\n")
mod.scan("x.env.example", b"AGENT_TOKEN=\nAGENT_ID=\n")
+184
View File
@@ -0,0 +1,184 @@
import pytest
from gti_discovery.catalog import FirebirdCatalog, MAX_SAMPLE_ROWS
from gti_discovery.identifiers import quote_firebird_identifier
from gti_discovery.firebird_queries import _ADAPTER_REGISTRATION_TOKEN, _execute_catalog, _register_approved_adapter
class RecordingAdapter:
read_only = True
def __init__(self, results):
self.results = results
self.calls = []
def execute(self, sql, params=()):
self.calls.append((sql, params))
return self.results[len(self.calls) - 1]
_register_approved_adapter(RecordingAdapter, _ADAPTER_REGISTRATION_TOKEN)
def make_catalog(sample_rows=None, relation='PRODUCTS', columns=('ID', 'EAN')):
adapter = RecordingAdapter([
[{'RELATION_NAME': relation, 'RELATION_ID': 1, 'SYSTEM_FLAG': 0}],
[{'RELATION_NAME': relation, 'FIELD_NAME': name} for name in columns],
sample_rows or [],
])
return FirebirdCatalog(adapter), adapter
@pytest.mark.parametrize('name', [
'PRODUCTS.ID', 'PRODUCTS; DROP TABLE X', 'PRODUCTS/* comment */',
'PRODUCTS -- comment', 'COUNT(*)', 'SP_RECALC(1)', '(SELECT 1)',
'PRODUCTS\nID',
])
def test_identifier_quoting_rejects_sql_shaped_input(name):
with pytest.raises(ValueError):
quote_firebird_identifier(name)
def test_identifier_quoting_doubles_embedded_quotes():
assert quote_firebird_identifier('WEIRD"NAME') == '"WEIRD""NAME"'
def test_sampling_accepts_catalog_issued_quoted_names():
catalog, adapter = make_catalog(
relation='WEIRD"TABLE', columns=('WEIRD"FIELD',)
)
catalog.sample_relation_rows('WEIRD"TABLE', ['WEIRD"FIELD'], 1)
assert adapter.calls[-1][0] == 'SELECT FIRST 1 "WEIRD""FIELD" FROM "WEIRD""TABLE"'
@pytest.mark.parametrize('relation', ['UNKNOWN', 'PRODUCTS; SELECT 1', 'SP_RECALC(1)'])
def test_sampling_rejects_unknown_or_procedure_like_relation(relation):
catalog, adapter = make_catalog()
with pytest.raises(ValueError):
catalog.sample_relation_rows(relation, ['ID'], 1)
assert not any('FIRST' in sql for sql, _ in adapter.calls)
@pytest.mark.parametrize('column', ['UNKNOWN', 'ID, PASSWORD', 'COUNT(*)', 'ID/*x*/'])
def test_sampling_rejects_unknown_or_expression_column(column):
catalog, adapter = make_catalog()
with pytest.raises(ValueError):
catalog.sample_relation_rows('PRODUCTS', [column], 1)
assert not any('FIRST' in sql for sql, _ in adapter.calls)
@pytest.mark.parametrize('limit', [0, -1, MAX_SAMPLE_ROWS + 1, '1', 1.0, True])
def test_sampling_rejects_invalid_or_oversized_limit(limit):
catalog, adapter = make_catalog()
with pytest.raises(ValueError):
catalog.sample_relation_rows('PRODUCTS', ['ID'], limit)
assert not any('FIRST' in sql for sql, _ in adapter.calls)
def test_sampling_returns_only_sanitized_scalar_evidence():
catalog, _ = make_catalog([{
'ID': 7,
'DESCRIPTION': 'widget',
'BLOB_DATA': b'binary',
'PASSWORD': 'secret',
'CUSTOMER_NAME': 'Customer One',
'NOT_SCALAR': {'nested': 'value'},
'OVERSIZED': 'x' * 2048,
}])
result = catalog.sample_relation_rows('PRODUCTS', ['ID'], 1)
assert result == [{
'ID': 7,
'DESCRIPTION': 'widget',
'BLOB_DATA': '<redacted>',
'OVERSIZED': '<redacted>',
}]
def test_sampling_driver_receives_generated_internal_sql_not_operation_key():
catalog, adapter = make_catalog()
catalog.sample_relation_rows('PRODUCTS', ['ID'], 1)
sql, params = adapter.calls[-1]
assert sql == 'SELECT FIRST 1 "ID" FROM "PRODUCTS"'
assert params == ()
assert 'DROP' not in sql.upper()
def test_catalog_identity_snapshots_are_immutable_and_detached():
relation_rows = [{'RELATION_NAME': 'PRODUCTS', 'RELATION_ID': 1, 'SYSTEM_FLAG': 0}]
field_rows = [{'RELATION_NAME': 'PRODUCTS', 'FIELD_NAME': 'ID'}]
adapter = RecordingAdapter([relation_rows, field_rows, []])
catalog = FirebirdCatalog(adapter)
relations = catalog.list_relations()
fields = catalog.list_relation_fields()
relation_rows[0]['RELATION_NAME'] = 'INJECTED'
field_rows[0]['FIELD_NAME'] = 'INJECTED'
with pytest.raises(TypeError):
relations[0]['RELATION_NAME'] = 'INJECTED'
with pytest.raises(AttributeError):
relations.append({'RELATION_NAME': 'INJECTED'})
with pytest.raises(TypeError):
fields[0]['FIELD_NAME'] = 'INJECTED'
catalog.sample_relation_rows('PRODUCTS', ['ID'], 1)
assert adapter.calls[-1][0] == 'SELECT FIRST 1 "ID" FROM "PRODUCTS"'
@pytest.mark.parametrize('params', [
('PRODUCTS; DROP TABLE X', ('ID',), 1),
('PRODUCTS', ('ID, PASSWORD',), 1),
('PRODUCTS', ('ID',), 0),
('PRODUCTS', ('ID',), MAX_SAMPLE_ROWS + 1),
('PRODUCTS', ('ID',), True),
('PRODUCTS', ('ID',), '1'),
])
def test_private_sampling_executor_rejects_invalid_structured_parameters(params):
adapter = RecordingAdapter([])
with pytest.raises(ValueError):
_execute_catalog(adapter, 'sample_relation', params)
assert adapter.calls == []
def test_private_sampling_executor_rejects_complete_sql_as_operation_input():
adapter = RecordingAdapter([])
with pytest.raises(ValueError):
_execute_catalog(adapter, 'SELECT FIRST 1 "ID" FROM "PRODUCTS"')
assert adapter.calls == []
def test_private_sampling_executor_rejects_unknown_valid_identifiers_without_snapshot_capability():
adapter = RecordingAdapter([])
with pytest.raises(ValueError):
_execute_catalog(adapter, 'sample_relation', ('UNKNOWN', ('ID',), 1))
assert adapter.calls == []
def test_consumer_cannot_bypass_snapshot_authorization_through_run_catalog():
catalog, adapter = make_catalog()
assert not hasattr(catalog, '_run_catalog')
with pytest.raises(AttributeError):
catalog._run_catalog('sample_relation', ('UNKNOWN', ('ID',), 1))
with pytest.raises(ValueError):
catalog._run_fixed_catalog('sample_relation')
assert adapter.calls == []
+45
View File
@@ -0,0 +1,45 @@
import importlib.util
import sys
from pathlib import Path
ROOT = Path(__file__).parents[1]
spec = importlib.util.spec_from_file_location("sgv_capabilities", ROOT / "scripts" / "sgv_capabilities.py")
cap = importlib.util.module_from_spec(spec); sys.modules[spec.name] = cap; spec.loader.exec_module(cap)
def columns(**over):
base = {f: 37 for f in cap.WRITE_TARGETS}
base.update({"ALIQUOTA": 16, "IPI": 27, cap.ORIGIN_FIELD: 8})
base.update(over)
return base
def test_capabilities_are_column_driven_not_version_driven():
caps = cap.detect_capabilities(columns())
assert all(caps.values()) and len(caps) == 15
missing = {k: v for k, v in columns().items() if k not in ("CEST", "COD_ORIGEM_MERCADORIA")}
caps = cap.detect_capabilities(missing)
assert caps["supports_cest"] is False and caps["supports_origin"] is False and caps["supports_ncm"] is True
def test_type_incompatibility_disables_capability():
assert cap.detect_capabilities(columns(IPI=37))["supports_ipi"] is False # text column cannot hold a numeric IPI rate
assert cap.detect_capabilities(columns(CST_ENTRADA=8))["supports_cst_entry"] is True # SMALLINT/INTEGER codes accepted
def test_schema_classification():
ok = cap.field_compatibility(columns())
assert cap.classify_schema(True, True, "abc", "abc", ok) == "REFERENCE_SCHEMA_MATCH"
assert cap.classify_schema(True, True, "xyz", "abc", ok) == "COMPATIBLE_WITH_VARIATIONS" # different hash is not a failure
assert cap.classify_schema(False, True, "abc", "abc", ok) == "INCOMPATIBLE"
bad = cap.field_compatibility({k: v for k, v in columns().items() if k != "CFOP"})
assert cap.classify_schema(True, True, "xyz", "abc", bad) == "INCOMPATIBLE"
no_origin = cap.field_compatibility({k: v for k, v in columns().items() if k != cap.ORIGIN_FIELD})
assert cap.classify_schema(True, True, "xyz", "abc", no_origin) == "COMPATIBLE_WITH_VARIATIONS" # origin is optional (only PROBABLE)
def test_segment_from_evidence_not_filename():
seg, _, ev = cap.infer_segment({"restaurant_tables_rows": 700, "kitchen_products": 200, "recipes": 0, "service_orders": 0, "vehicle_products": 0, "grade_items": 0, "nfce_items": 0}, "")
assert seg == "RESTAURANT" and ev
assert cap.infer_segment({}, "MM MODAS")[0] == "APPAREL"
assert cap.infer_segment({}, "")[0] == "UNKNOWN"
+52
View File
@@ -0,0 +1,52 @@
import importlib.util
import json
import struct
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).parents[1]
spec = importlib.util.spec_from_file_location("extract_sgv_dfm_bindings", ROOT / "scripts" / "extract_sgv_dfm_bindings.py")
dfm = importlib.util.module_from_spec(spec); sys.modules[spec.name] = dfm; spec.loader.exec_module(dfm)
def s(text):
b = text.encode("latin-1"); return bytes([len(b)]) + b
def prop_str(name, value):
return s(name) + b"\x06" + s(value)
def component(cls, name, props=b"", kids=b""):
return s(cls) + s(name) + props + b"\x00" + kids + b"\x00"
def test_parser_reads_nested_controls_datafields_and_collections():
grid_cols = s("Columns") + b"\x0e" + b"\x01" + s("Visible") + b"\x09" + b"\x00" + b"\x01" + s("FieldName") + b"\x06" + s("QTDE") + b"\x00" + b"\x00"
edit = component("TDBEdit", "DBEdit72", prop_str("DataField", "CFOP_ENTRADA") + prop_str("DataSource", "dsproduto2"))
group = component("TGroupBox", "FlatGroupBox5", prop_str("Caption", "Nota Fiscal de Entrada"), edit)
grid = component("TDBGrid", "Grid1", grid_cols)
form = b"TPF0" + component("Tfrm", "frm", b"", group + grid)
r = dfm.R(form, 4); tree = r.component()
out = []; dfm.walk(tree, [], out)
ctl = next(c for c in out if c["DataField"] == "CFOP_ENTRADA")
assert ctl["name"] == "DBEdit72" and ctl["DataSource"] == "dsproduto2" and any("Nota Fiscal de Entrada" in p for p in ctl["path"])
assert r.p == len(form) # whole stream consumed, collections included
def test_ui_map_uses_dfm_evidence_and_keeps_policies():
p = ROOT / "docs" / "SGV_PRODUCT_FISCAL_UI_MAP_V1.json"
if not p.exists():
pytest.skip("map not generated")
d = json.loads(p.read_text(encoding="utf-8"))
assert d["transactional_table_writes_allowed"] is False
by = {f["ui_label"]: f for f in d["fields"]}
assert by["Pis Entrada"]["field"] == "COD_ENTRADA_PIS" and by["Cofins Saída"]["field"] == "COD_SAIDA_COFINS"
assert by["NCM"]["field"] == "CLASSIFICACAO_FISCAL" and by["Seletor ao lado do CEST (T/I/N/S)"]["field"] == "SITUACAO_TRIBUTARIA"
assert by["Código da origem da Mercadoria (0/1/2)"]["classification"] == "CONFIRMED_STRONG" and by["Código da origem da Mercadoria (0/1/2)"]["write_target"]
assert by["CSOSN Informações - Significados"]["classification"] == "NO_MATCH" and not by["CSOSN Informações - Significados"]["write_target"]
for f in d["fields"]:
if f["write_target"]:
assert f["classification"] == "CONFIRMED_STRONG" and f["table"] == "C000025" and f["evidence_basis"]
+72
View File
@@ -0,0 +1,72 @@
import importlib.util
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).parents[1]
def load(name, rel):
sys.path.insert(0, str(ROOT / "agent" / "src")); sys.path.insert(0, str(ROOT / "src"))
spec = importlib.util.spec_from_file_location(name, ROOT / rel)
mod = importlib.util.module_from_spec(spec); sys.modules[name] = mod; spec.loader.exec_module(mod)
return mod
snap = load("sgv_product_snapshot", "scripts/sgv_product_snapshot.py")
audit = load("audit_firebird_database", "scripts/audit_firebird_database.py")
executor = load("firebird_write_executor", "scripts/firebird_write_executor.py")
def test_diff_lists_changed_columns_and_flags_technical_ones():
before = {"CFOP": "5102", "CSOSN": "102", "ULTIMA_ALTERACAO": "2026-01-01"}
after = {"CFOP": "5405", "CSOSN": "102", "ULTIMA_ALTERACAO": "2026-02-02"}
d = {c["FIELD"]: c for c in snap.compute_diff(before, after)}
assert d["CFOP"]["BEFORE"] == "5102" and d["CFOP"]["AFTER"] == "5405" and not d["CFOP"]["technical"]
assert d["ULTIMA_ALTERACAO"]["technical"] and "CSOSN" not in d
def test_ledger_confirms_only_single_column_changes(tmp_path, monkeypatch):
monkeypatch.setattr(snap, "LEDGER", tmp_path / "ledger.json")
ok = snap.record("CFOP saída", "1", [{"FIELD": "CFOP", "BEFORE": "a", "AFTER": "b", "technical": False}, {"FIELD": "ULTIMA_ALTERACAO", "BEFORE": 1, "AFTER": 2, "technical": True}])
two = snap.record("X", "1", [{"FIELD": "A", "BEFORE": 1, "AFTER": 2, "technical": False}, {"FIELD": "B", "BEFORE": 1, "AFTER": 2, "technical": False}])
none = snap.record("Y", "1", [])
assert ok["confirmed"] and not two["confirmed"] and not none["confirmed"]
assert len(json.loads((tmp_path / "ledger.json").read_text(encoding="utf-8"))) == 3
def test_company_context_is_read_generically_and_fails_closed():
from gti_fiscal_agent.firebird_reader import parse_company_context
assert parse_company_context([("sp", "3")]) == {"uf": "SP", "crt": "3"}
for bad in ([], [("BA", "1"), ("SP", "1")], [("", "1")], [("BA", "9")]):
with pytest.raises(ValueError):
parse_company_context(bad)
def test_audit_helpers():
assert audit.ncm_bucket("22030000") == "eight_digits" and audit.ncm_bucket(" ") == "blank" and audit.ncm_bucket("2203.00") == "other_format"
c = audit.compat({"CFOP": 37, "IPI": 27, "CST": 8})
assert c["CFOP"] == "OK" and c["IPI"] == "OK" and c["CST"].startswith("TYPE_DIFFERS") and c["CSOSN"] == "MISSING"
@pytest.mark.parametrize("table", ["NFCE", "NFCE_ITENS", "C000032", "C000061", "C000062", "C000088", "INVENTARIO"])
def test_transactional_tables_are_never_writable(table):
with pytest.raises(executor.WriteRefused):
executor.assert_allowed_field(table, "CFOP")
def test_ui_map_write_targets_have_evidence_and_transactional_writes_forbidden():
p = ROOT / "docs" / "SGV_PRODUCT_FISCAL_UI_MAP_V1.json"
if not p.exists():
pytest.skip("map not generated")
d = json.loads(p.read_text(encoding="utf-8"))
assert d["transactional_table_writes_allowed"] is False
for f in d["fields"]:
assert f["table"] == "C000025"
if f["write_target"]:
assert f["confirmed_by_ui_diff"] or f["evidence_basis"]
assert f["field"]
if f.get("classification") in ("PROBABLE", "AMBIGUOUS", "NO_MATCH"):
assert not f["write_target"]