GTI_GERENCIADOR_FISCAL/fiscal/tests/test_sampling.py

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 == []