38 lines
2.1 KiB
Python
38 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise gateway imports and its migration command using disposable state."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1]/'resources/scripts/napseer_mcp_server.py'
|
|
|
|
with tempfile.TemporaryDirectory(prefix='gateway-private-state-') as directory:
|
|
root=Path(directory)/'workspace'; root.mkdir()
|
|
data=Path(directory)/'data'
|
|
env=dict(os.environ,NAPSEER_USER_DATA_DIR=str(data),NAPSEER_PROJECT_ROOT=str(root),NAPSEER_TELEMETRY='0')
|
|
for name in ('NAPSEER_TOKEN','NAPSEER_REFRESH_TOKEN','NAPSEER_API_KEY','NAPSEER_AUTH_FILE'):
|
|
env.pop(name,None)
|
|
code="import sys; sys.path.insert(0,sys.argv[1]); import napseer_mcp_server as m; import json; print(json.dumps({'state':str(m.AUTH_DIR),'auth':str(m.AUTH_PATH),'key':str(m.DEFAULT_KEY_PATH)}))"
|
|
def paths():
|
|
result=subprocess.run([sys.executable,'-c',code,str(SCRIPT.parent)],cwd=root,env=env,capture_output=True,text=True,timeout=10)
|
|
assert result.returncode==0,'gateway must load the private-state dependency'
|
|
return {k:Path(v) for k,v in json.loads(result.stdout).items()}
|
|
resolved=paths()
|
|
assert all(path.is_relative_to(data) and not path.is_relative_to(root) for path in resolved.values())
|
|
legacy=root/'.napseer';legacy.mkdir()
|
|
(legacy/'auth.json').write_text(json.dumps({'token':'synthetic-worker-token','worker_id':'test-worker'}))
|
|
(legacy/'id_ed25519').write_bytes(b'synthetic-signing-key')
|
|
result=subprocess.run([sys.executable,str(SCRIPT),'auth','migrate'],cwd=root,env=env,capture_output=True,text=True,timeout=10)
|
|
assert result.returncode==0,'gateway migration must complete without enrollment'
|
|
assert json.loads(result.stdout)['status']=='migrated'
|
|
assert 'synthetic-worker-token' not in result.stdout+result.stderr
|
|
assert json.loads(resolved['auth'].read_text())['worker_id']=='test-worker'
|
|
assert resolved['key'].read_bytes()==b'synthetic-signing-key'
|
|
assert not list(legacy.iterdir())
|
|
assert paths()==resolved
|
|
|
|
print('ok: gateway private state and migration stay outside the checkout')
|