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", )