275 lines
10 KiB
Python
275 lines
10 KiB
Python
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)
|