763 lines
31 KiB
Python
763 lines
31 KiB
Python
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"
|