feat(notion-writer): ntn_files upload + preflight
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
#!/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 run_all():
|
||||
tests = [
|
||||
test_preflight_missing_ntn,
|
||||
test_preflight_returns_workspace,
|
||||
test_upload_returns_id,
|
||||
test_upload_failure_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()
|
||||
Reference in New Issue
Block a user