Files
Andrew Yim 5c904756ab fix(ntn-files): propagate NOTION_API_KEY into ntn subprocesses
Force ntn to authenticate as the same integration that writes the page by
injecting NOTION_API_TOKEN=$NOTION_API_KEY into every ntn subprocess env.
Notion scopes file uploads to the creating integration, so a mismatch caused
"Could not find file_upload with ID …" when attaching uploaded images.

Added test_upload_passes_token_env (9 total in test_ntn_files.py) and updated
code/CLAUDE.md to reflect that a separate ntn login is no longer required.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 11:13:31 +09:00

153 lines
4.9 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 test_upload_passes_token_env():
_reset_cache()
fake = subprocess.CompletedProcess(args=[], returncode=0,
stdout="ID123\tphoto.png\tuploaded\n", stderr="")
with mock.patch.dict("os.environ", {"NOTION_API_KEY": "tok-abc"}, clear=False), \
mock.patch("builtins.open", mock.mock_open(read_data=b"x")), \
mock.patch("subprocess.run", return_value=fake) as m:
ntn_files.upload(Path("/tmp/p.png"))
passed_env = m.call_args.kwargs["env"]
assert passed_env["NOTION_API_TOKEN"] == "tok-abc"
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,
test_upload_passes_token_env,
]
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()