161 lines
7.5 KiB
Python
161 lines
7.5 KiB
Python
"""Real Git worktrees and worker processes exercise private-state ownership."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / 'resources/scripts'
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
from napseer_credentials import CredentialStore, atomic_json, read_json
|
|
from test_user_credentials import provider, worker, outcome
|
|
|
|
|
|
@pytest.fixture
|
|
def worktrees(tmp_path):
|
|
main, other = tmp_path/'main', tmp_path/'other'
|
|
main.mkdir()
|
|
def git(*args):
|
|
subprocess.run(['git', '-C', str(main), *args], check=True, capture_output=True)
|
|
git('init', '-q')
|
|
atomic_json(main/'.napseer/project.json', {'schema':'napseer.project.v1',
|
|
'base_url':'https://api.example.test', 'project_id':'33333333-3333-3333-3333-333333333333',
|
|
'project_slug':'shared'})
|
|
git('add', '.napseer/project.json')
|
|
git('-c', 'user.name=Test', '-c', 'user.email=test@example.test', 'commit', '-qm', 'locator')
|
|
git('worktree', 'add', '-qb', 'other', str(other))
|
|
return main, other
|
|
|
|
|
|
def test_worker_keeps_private_paths_outside_checkout_and_stable_from_subdirectory(worktrees):
|
|
main, other = worktrees
|
|
sub = main/'nested'; sub.mkdir()
|
|
expression = "{'state':str(m.local_state_dir()),'key':str(m.DEFAULT_KEY_PATH),'locator':str(m.project_locator_path())}"
|
|
a,b,c = [outcome(worker(root, expression))['result'] for root in (main,other,sub)]
|
|
assert not Path(a['state']).is_relative_to(main)
|
|
assert not Path(b['state']).is_relative_to(other)
|
|
assert a['state'] == c['state']
|
|
assert a['state'] != b['state']
|
|
assert a['key'] == b['key']
|
|
assert a['locator'] == str(main/'.napseer/project.json')
|
|
assert c['locator'] == a['locator']
|
|
|
|
|
|
def test_project_login_is_shared_across_worktrees_and_refreshes_once(worktrees, provider):
|
|
main, other = worktrees
|
|
url, state = provider
|
|
first, second = [CredentialStore(root/'.napseer') for root in (main,other)]
|
|
first.login({'base_url':url,'token':'old-access','refresh_token':'old-refresh',
|
|
'account_mode':'operator_project','oauth_client_id':'nap-cli'}, project=True)
|
|
assert first.project == second.project
|
|
assert first.selection != second.selection
|
|
assert not first.selection.is_relative_to(main)
|
|
requests=[worker(root, "m.request_json('GET','/v1/projects')") for root in (main,other)]
|
|
assert all(outcome(p)['ok'] for p in requests)
|
|
assert state['refreshes'] == 1
|
|
assert outcome(worker(other, "m.request_json('GET','/v1/projects')"))['ok']
|
|
assert state['refreshes'] == 1
|
|
|
|
|
|
def test_migration_preserves_keys_and_locator_and_refuses_conflicting_destination(worktrees):
|
|
from napseer_credentials import WorkspacePaths, CredentialError
|
|
main, _ = worktrees
|
|
paths = WorkspacePaths(main)
|
|
old = main/'.napseer'
|
|
locator=(old/'project.json').read_bytes()
|
|
(old/'vault.local-key').write_bytes(b'synthetic-vault-key\n')
|
|
(old/'vault.json').write_bytes(b'{"synthetic":"ciphertext"}\n')
|
|
(old/'id_ed25519').write_bytes(b'synthetic-signing-key\n')
|
|
atomic_json(old/'auth.json', {'token':'synthetic-token','ssh_key_path':str(old/'id_ed25519')})
|
|
paths.state.mkdir(parents=True)
|
|
(paths.state/'vault.local-key').write_bytes(b'conflicting-key')
|
|
with pytest.raises(CredentialError): paths.migrate()
|
|
assert (old/'vault.local-key').read_bytes() == b'synthetic-vault-key\n'
|
|
assert (old/'auth.json').exists()
|
|
(paths.state/'vault.local-key').unlink()
|
|
paths.migrate()
|
|
assert (paths.state/'vault.local-key').read_bytes() == b'synthetic-vault-key\n'
|
|
assert paths.signing_key.read_bytes() == b'synthetic-signing-key\n'
|
|
assert read_json(CredentialStore(old).project)['ssh_key_path'] == str(paths.signing_key)
|
|
assert (old/'project.json').read_bytes() == locator
|
|
assert set(p.name for p in old.iterdir()) == {'project.json'}
|
|
paths.migrate() # retry after a completed migration is harmless
|
|
assert paths.signing_key.stat().st_mode & 0o777 == 0o600
|
|
|
|
|
|
def test_other_repositories_do_not_share_project_override(tmp_path):
|
|
first=CredentialStore(tmp_path/'one/.napseer')
|
|
second=CredentialStore(tmp_path/'two/.napseer')
|
|
first.login({'token':'synthetic-project-token'}, project=True)
|
|
assert first.project != second.project
|
|
assert not second.read().get('token')
|
|
|
|
|
|
def test_linked_worktree_move_keeps_external_state_identity(worktrees):
|
|
from napseer_credentials import WorkspacePaths
|
|
main, other = worktrees
|
|
before = WorkspacePaths(other)
|
|
moved = other.with_name('moved')
|
|
subprocess.run(['git','-C',str(main),'worktree','move',str(other),str(moved)], check=True, capture_output=True)
|
|
after = WorkspacePaths(moved)
|
|
assert before.state == after.state
|
|
assert before.project_credentials == after.project_credentials
|
|
|
|
|
|
def test_concurrent_worktree_signing_key_creation_is_serialized(worktrees):
|
|
main, other = worktrees
|
|
processes = [worker(root, 'str(m.ensure_local_files())') for root in (main,other)]
|
|
results = [outcome(p) for p in processes]
|
|
assert all(r['ok'] for r in results)
|
|
assert results[0]['result'] == results[1]['result']
|
|
key = Path(results[0]['result'])
|
|
assert key.is_file() and Path(str(key)+'.pub').is_file()
|
|
assert key.stat().st_mode & 0o777 == 0o600
|
|
assert all(not (root/'.napseer/id_ed25519').exists() for root in (main,other))
|
|
|
|
|
|
def test_interrupted_migration_preserves_originals_and_retries(worktrees, monkeypatch):
|
|
from napseer_credentials import WorkspacePaths
|
|
main, _ = worktrees
|
|
paths=WorkspacePaths(main)
|
|
for name in ('vault.local-key','vault.json'):(paths.legacy/name).write_bytes(b'synthetic-'+name.encode())
|
|
original_link=os.link
|
|
calls=[]
|
|
def interrupted(source,destination):
|
|
calls.append(destination)
|
|
if len(calls)==2:raise OSError('synthetic interrupted publication')
|
|
return original_link(source,destination)
|
|
monkeypatch.setattr(os,'link',interrupted)
|
|
with pytest.raises(OSError):paths.migrate()
|
|
assert all((paths.legacy/name).exists() for name in ('vault.local-key','vault.json'))
|
|
monkeypatch.setattr(os,'link',original_link)
|
|
paths.migrate()
|
|
assert all((paths.state/name).read_bytes()==b'synthetic-'+name.encode() for name in ('vault.local-key','vault.json'))
|
|
|
|
|
|
def test_cli_migration_reports_running_gateway_and_recovers(worktrees):
|
|
main, _ = worktrees
|
|
atomic_json(main/'.napseer/gateway.json', {'pid':os.getpid()})
|
|
key=main/'.napseer/vault.local-key';key.write_text('synthetic-key')
|
|
env=dict(os.environ,NAPSEER_PROJECT_ROOT=str(main),NAPSEER_TELEMETRY='0')
|
|
command=[sys.executable,str(SCRIPTS/'napseer_mcp_server.py'),'auth','migrate']
|
|
result=subprocess.run(command,cwd=main,env=env,capture_output=True,text=True,timeout=15)
|
|
assert result.returncode!=0 and 'Stop the running' in result.stderr
|
|
assert 'synthetic-key' not in result.stderr and key.exists()
|
|
atomic_json(main/'.napseer/gateway.json', {'pid':0})
|
|
result=subprocess.run(command,cwd=main,env=env,capture_output=True,text=True,timeout=15)
|
|
assert result.returncode==0
|
|
report=json.loads(result.stdout)
|
|
assert report['status']=='migrated'
|
|
assert (Path(report['state_dir'])/'vault.local-key').read_text()=='synthetic-key'
|
|
assert not key.exists()
|
|
|
|
|
|
def test_private_state_cannot_be_redirected_into_checkout(worktrees, monkeypatch):
|
|
from napseer_credentials import WorkspacePaths, CredentialError
|
|
main,_=worktrees
|
|
monkeypatch.setenv('NAPSEER_USER_DATA_DIR',str(main/'private-state'))
|
|
with pytest.raises(CredentialError,match='outside'):WorkspacePaths(main)
|