68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""Manually set C000025.IPI back to NULL for one test product, to simulate ERP
|
|
drift for the incremental auto-apply test. Guarded the same way as any other
|
|
write here (path + allowlist + schema), executed once, by hand, outside the
|
|
normal Agent flow — this is test-fixture setup, not something the Agent does.
|
|
"""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def load_dotenv(path):
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
os.environ[key.strip()] = value.strip()
|
|
|
|
|
|
def load_executor():
|
|
spec = importlib.util.spec_from_file_location("firebird_write_executor", ROOT / "scripts" / "firebird_write_executor.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def main():
|
|
code = sys.argv[1] if len(sys.argv) > 1 else "000026"
|
|
load_dotenv(ROOT / ".env")
|
|
executor = load_executor()
|
|
database_path = os.environ["FIREBIRD_DATABASE_PATH"]
|
|
executor.assert_authorized_write_path(database_path)
|
|
executor.assert_allowed_field("C000025", "IPI")
|
|
|
|
import fdb
|
|
conn = fdb.connect(
|
|
database=database_path,
|
|
user=os.environ["FB_USER"],
|
|
password=os.environ["FB_PASSWORD"],
|
|
fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"),
|
|
)
|
|
try:
|
|
cur = conn.cursor()
|
|
executor.assert_schema_ok(cur, "C000025", "IPI", None)
|
|
cur.execute("SELECT IPI FROM C000025 WHERE CODIGO = ?", (code,))
|
|
row = cur.fetchone()
|
|
before = row[0] if row else None
|
|
cur.execute("UPDATE C000025 SET IPI = NULL WHERE CODIGO = ? AND IPI = ?", (code, before))
|
|
rowcount = cur.rowcount
|
|
if rowcount != 1:
|
|
conn.rollback()
|
|
raise SystemExit(f"expected rowcount 1, got {rowcount}")
|
|
conn.commit()
|
|
cur.execute("SELECT IPI FROM C000025 WHERE CODIGO = ?", (code,))
|
|
after = cur.fetchone()[0]
|
|
print(json.dumps({"product_code": code, "before": before, "after": after, "rowcount": rowcount}, indent=2))
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|