258 lines
8.4 KiB
Python
258 lines
8.4 KiB
Python
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",
|
|
)
|