140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for ntn_files.py — run with `python test_ntn_files.py`."""
|
|
|
|
import sys
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
import ntn_files
|
|
from ntn_files import NtnUploadError
|
|
|
|
|
|
def _reset_cache():
|
|
ntn_files._WORKSPACE = None
|
|
|
|
|
|
def test_preflight_missing_ntn():
|
|
_reset_cache()
|
|
with mock.patch("shutil.which", return_value=None):
|
|
try:
|
|
ntn_files.preflight()
|
|
assert False, "expected NtnUploadError"
|
|
except NtnUploadError as e:
|
|
assert "ntn" in str(e).lower()
|
|
|
|
|
|
def test_preflight_returns_workspace():
|
|
_reset_cache()
|
|
fake = subprocess.CompletedProcess(
|
|
args=[], returncode=0,
|
|
stdout='{"bot":{"workspace_name":"D.Intelligence",'
|
|
'"workspace_id":"ws-123"}}',
|
|
stderr="",
|
|
)
|
|
with mock.patch("shutil.which", return_value="/usr/bin/ntn"), \
|
|
mock.patch("subprocess.run", return_value=fake):
|
|
info = ntn_files.preflight()
|
|
assert info["workspace_name"] == "D.Intelligence"
|
|
assert info["workspace_id"] == "ws-123"
|
|
|
|
|
|
def test_upload_returns_id():
|
|
_reset_cache()
|
|
fake = subprocess.CompletedProcess(
|
|
args=[], returncode=0,
|
|
stdout="43833259-72ae-404e-8441-b6577f3159b4\tphoto.png\tuploaded\n",
|
|
stderr="",
|
|
)
|
|
# upload() opens the file before subprocess.run; mock open so the file
|
|
# need not exist (subprocess.run is mocked and never reads the handle).
|
|
with mock.patch("builtins.open", mock.mock_open(read_data=b"x")), \
|
|
mock.patch("subprocess.run", return_value=fake):
|
|
upload_id = ntn_files.upload(Path("/tmp/photo.png"))
|
|
assert upload_id == "43833259-72ae-404e-8441-b6577f3159b4"
|
|
|
|
|
|
def test_upload_failure_raises():
|
|
_reset_cache()
|
|
fake = subprocess.CompletedProcess(
|
|
args=[], returncode=1, stdout="", stderr="boom: invalid file",
|
|
)
|
|
with mock.patch("builtins.open", mock.mock_open(read_data=b"x")), \
|
|
mock.patch("subprocess.run", return_value=fake):
|
|
try:
|
|
ntn_files.upload(Path("/tmp/bad.png"))
|
|
assert False, "expected NtnUploadError"
|
|
except NtnUploadError as e:
|
|
assert e.path == "/tmp/bad.png"
|
|
assert "boom" in e.stderr
|
|
|
|
|
|
def _img(url):
|
|
return {"object": "block", "type": "image",
|
|
"image": {"type": "external", "external": {"url": url}}}
|
|
|
|
|
|
def test_materialize_remote_untouched():
|
|
blocks = [_img("https://ex.com/a.png")]
|
|
out = ntn_files.materialize_local_media(
|
|
blocks, Path("/tmp"), upload_fn=lambda p: "SHOULD_NOT_RUN")
|
|
assert out[0]["image"]["type"] == "external"
|
|
|
|
|
|
def test_materialize_local_uploaded(tmp_path=None):
|
|
import tempfile, os
|
|
d = Path(tempfile.mkdtemp())
|
|
(d / "x.png").write_bytes(b"fake")
|
|
blocks = [_img("x.png")]
|
|
out = ntn_files.materialize_local_media(
|
|
blocks, d, upload_fn=lambda p: "UP123")
|
|
assert out[0]["image"]["type"] == "file_upload"
|
|
assert out[0]["image"]["file_upload"]["id"] == "UP123"
|
|
|
|
|
|
def test_materialize_nested_children():
|
|
import tempfile
|
|
d = Path(tempfile.mkdtemp())
|
|
(d / "y.png").write_bytes(b"fake")
|
|
toggle = {"object": "block", "type": "toggle",
|
|
"toggle": {"rich_text": [], "children": [_img("y.png")]}}
|
|
out = ntn_files.materialize_local_media(
|
|
[toggle], d, upload_fn=lambda p: "UPNESTED")
|
|
child = out[0]["toggle"]["children"][0]
|
|
assert child["image"]["file_upload"]["id"] == "UPNESTED"
|
|
|
|
|
|
def test_materialize_missing_file_raises():
|
|
blocks = [_img("nope.png")]
|
|
try:
|
|
ntn_files.materialize_local_media(
|
|
blocks, Path("/tmp"), upload_fn=lambda p: "x")
|
|
assert False, "expected NtnUploadError"
|
|
except NtnUploadError as e:
|
|
assert "nope.png" in str(e) or "nope.png" in e.path
|
|
|
|
|
|
def run_all():
|
|
tests = [
|
|
test_preflight_missing_ntn,
|
|
test_preflight_returns_workspace,
|
|
test_upload_returns_id,
|
|
test_upload_failure_raises,
|
|
test_materialize_remote_untouched,
|
|
test_materialize_local_uploaded,
|
|
test_materialize_nested_children,
|
|
test_materialize_missing_file_raises,
|
|
]
|
|
for t in tests:
|
|
print(f"\n{t.__name__}")
|
|
t()
|
|
print("\n" + "=" * 50)
|
|
print(f"✅ All {len(tests)} tests passed")
|
|
print("=" * 50)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_all()
|