"""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 # Released CLI 0.4.0 source revision (pinned by the Napseer deploy workflow). RELEASED_0_4_0 = 'c427e2432ac85038a4f00b2b888bbdb7c851ada1' @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) def test_root_tools_file_resolves_like_locator_and_is_invisible_to_migration(worktrees, monkeypatch): from napseer_credentials import WorkspacePaths main, _ = worktrees monkeypatch.delenv('NAPSEER_TOOL_PROFILES', raising=False) tools = main/'.napseer-tools.json' committed = b'{"schema": "napseer.tools.v1", "tool_profiles": ["workflow"]}\n' tools.write_bytes(committed) sub = main/'nested'; sub.mkdir() # A worker started below the project root finds the file at the root that # owns .napseer/project.json, the same way it finds the locator. expression = ("{'path':str(m.project_tools_path()),'profiles':sorted(m.tool_profile_configuration()[0])," "'source':m.tool_profile_configuration()[1]," "'pending':sorted(p.name for p in m.WORKSPACE.migration_sources())}") result = outcome(worker(sub, expression))['result'] assert result == {'path': str(tools), 'profiles': ['workflow'], 'source': 'tools_file', 'pending': []} assert set(p.name for p in (main/'.napseer').iterdir()) <= {'project.json', '.gitignore'} paths = WorkspacePaths(main) assert paths.migration_sources() == [] paths.require_migrated() paths.migrate() assert tools.read_bytes() == committed def test_root_tools_file_is_invisible_to_released_0_4_0_migration_check(worktrees, tmp_path): """CLI 0.4.0 treats any extra .napseer/ entry as private state to migrate.""" import importlib.util released = subprocess.run( ['git', '-C', str(SCRIPTS.parent.parent), 'show', f'{RELEASED_0_4_0}:resources/scripts/napseer_credentials.py'], capture_output=True) if released.returncode != 0: pytest.skip(f'released 0.4.0 source {RELEASED_0_4_0} is not in this checkout') source = tmp_path/'napseer_credentials_0_4_0.py' source.write_bytes(released.stdout) spec = importlib.util.spec_from_file_location('napseer_credentials_0_4_0', source) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) main, _ = worktrees (main/'.napseer-tools.json').write_text('{"schema": "napseer.tools.v1", "tool_profiles": ["workflow"]}\n') paths = module.WorkspacePaths(main) assert paths.migration_sources() == [] paths.require_migrated() # Control: the same file inside .napseer/ would block 0.4.0 clients. (main/'.napseer/tools.json').write_text('{}\n') assert [p.name for p in paths.migration_sources()] == ['tools.json'] with pytest.raises(module.CredentialError): paths.require_migrated()