mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-22 08:25:21 +08:00
947d8addea
* fix(box): retain outbox files when export fails * test(box): retain ownership evidence in CI logs * test(box): exercise LocalAgent export repair through real runtime and Docker Co-authored-by: multica-agent <[email protected]> * fix(box): complete bounded outbox recovery contract * feat(box): export explicitly selected files Co-authored-by: multica-agent <[email protected]> * test(box): exercise explicit export on real Docker Co-authored-by: multica-agent <[email protected]> * test(box): verify explicit exports with live model and Docker Co-authored-by: multica-agent <[email protected]> * fix(box): align explicit export SDK and integration contracts * docs(box): describe selected-file export contract --------- Co-authored-by: HaiYangBG1 <[email protected]> Co-authored-by: Chin Rock <[email protected]> Co-authored-by: multica-agent <[email protected]>
58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
import hashlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
from langbot.pkg.box.selected_files import selected_files_script
|
|
|
|
|
|
def run(paths, cleanup=None, limit=100):
|
|
return subprocess.run(
|
|
[
|
|
sys.executable,
|
|
'-c',
|
|
selected_files_script(list(map(str, paths)), max_file_bytes=limit, max_total_bytes=limit, cleanup=cleanup),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
def test_selected_read_and_cleanup(tmp_path):
|
|
a, b = tmp_path / 'a', tmp_path / 'b'
|
|
a.write_bytes(b'hello')
|
|
b.write_bytes(b'keep')
|
|
result = run([a])
|
|
assert result.returncode == 0, result.stderr
|
|
assert json.loads(result.stdout) == [{'name': str(a), 'b64': 'aGVsbG8='}]
|
|
assert a.exists() and b.exists()
|
|
assert run([a], {str(a): hashlib.sha256(b'hello').hexdigest()}).returncode == 0
|
|
assert not a.exists() and b.read_bytes() == b'keep'
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['missing', 'symlink', 'directory', 'size', 'fifo'])
|
|
def test_failed_batch_retains_all(tmp_path, bad):
|
|
a, b = tmp_path / 'a', tmp_path / 'b'
|
|
a.write_bytes(b'hello')
|
|
if bad == 'symlink':
|
|
b.symlink_to(a)
|
|
elif bad == 'directory':
|
|
b.mkdir()
|
|
elif bad == 'size':
|
|
b.write_bytes(b'x' * 101)
|
|
elif bad == 'fifo':
|
|
import os
|
|
|
|
os.mkfifo(b)
|
|
result = run([a, b])
|
|
assert result.returncode != 0
|
|
assert a.read_bytes() == b'hello'
|
|
|
|
|
|
def test_changed_file_is_retained(tmp_path):
|
|
a = tmp_path / 'a'
|
|
a.write_bytes(b'changed')
|
|
assert run([a], {str(a): hashlib.sha256(b'old').hexdigest()}).returncode != 0
|
|
assert a.read_bytes() == b'changed'
|