| 1 | """Integration tests for witdb-bridge subprocess backend.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import subprocess |
| 7 | import tempfile |
| 8 | from pathlib import Path |
| 9 | |
| 10 | import pytest |
| 11 | |
| 12 | from pywitdb._bridge.client import bridge_available, resolve_bridge_executable |
| 13 | from pywitdb._bridge.db import open_bridge |
| 14 | |
| 15 | pytestmark = pytest.mark.skipif( |
| 16 | not bridge_available(), reason="witdb-bridge not built" |
| 17 | ) |
| 18 | |
| 19 | |
| 20 | def test_bridge_kv_roundtrip() -> None: |
| 21 | with tempfile.TemporaryDirectory() as tmp: |
| 22 | path = str(Path(tmp) / "bridge.witdb") |
| 23 | db = open_bridge(path, create=True) |
| 24 | try: |
| 25 | db.put(b"k", b"v") |
| 26 | assert db.get(b"k") == b"v" |
| 27 | assert db.delete(b"k") is True |
| 28 | assert db.get(b"k") is None |
| 29 | finally: |
| 30 | db.close() |
| 31 | |
| 32 | |
| 33 | def test_public_open_auto_fallback() -> None: |
| 34 | import pywitdb |
| 35 | |
| 36 | with tempfile.TemporaryDirectory() as tmp: |
| 37 | path = str(Path(tmp) / "via_auto.witdb") |
| 38 | with pywitdb.open(path, backend="auto") as db: |
| 39 | db.put(b"auto", b"1") |
| 40 | assert db.get(b"auto") == b"1" |
| 41 | |
| 42 | |
| 43 | def test_bridge_open_metadata() -> None: |
| 44 | exe = resolve_bridge_executable() |
| 45 | assert exe is not None |
| 46 | with tempfile.TemporaryDirectory() as tmp: |
| 47 | path = str(Path(tmp) / "meta.witdb") |
| 48 | proc = subprocess.Popen( |
| 49 | [str(exe)], |
| 50 | stdin=subprocess.PIPE, |
| 51 | stdout=subprocess.PIPE, |
| 52 | stderr=subprocess.PIPE, |
| 53 | text=True, |
| 54 | ) |
| 55 | assert proc.stdin and proc.stdout |
| 56 | proc.stdin.write( |
| 57 | json.dumps({"id": 1, "op": "open", "path": path, "create": True}) + "\n" |
| 58 | ) |
| 59 | proc.stdin.flush() |
| 60 | line = proc.stdout.readline() |
| 61 | proc.stdin.write(json.dumps({"id": 2, "op": "close"}) + "\n") |
| 62 | proc.stdin.flush() |
| 63 | proc.stdin.close() |
| 64 | proc.wait(timeout=10) |
| 65 | resp = json.loads(line) |
| 66 | assert resp["ok"] is True |
| 67 | assert resp["store"]["store_provider"] == "btree" |
| 68 | |