tegwick
3db3e7e297
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a087d8-d22d-7d11-8169-bf22a729dff1
63 lines
3 KiB
Python
63 lines
3 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from repo_manager.identifiers import ensure_missing_work_record_identifiers
|
|
from repo_manager.record_identity import scan_record_identities
|
|
|
|
|
|
def _sources(root: Path, *, status='completed', task_status='done',
|
|
task_id='NK-WP-0003-T08a', binding='22222222-2222-4222-8222-222222222222'):
|
|
archive = root / 'workplans' / 'archived' / 'NK-WP-0003.md'
|
|
archive.parent.mkdir(parents=True)
|
|
archive.write_text(
|
|
f'---\nid: NK-WP-0003\nstatus: {status}\n'
|
|
'state_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---\n'
|
|
f'\n```task\nid: {task_id}\nstatus: {task_status}\n'
|
|
f'state_hub_task_id: "{binding}"\n```\n'
|
|
)
|
|
live = root / 'workplans' / 'NK-WP-0099.md'
|
|
live.write_text('---\nid: NK-WP-0099\nstatus: active\n---\n'
|
|
'\n```task\nid: NK-WP-0099-T01\nstatus: todo\n```\n')
|
|
return archive, live
|
|
|
|
|
|
@pytest.mark.parametrize('status', ['finished', 'archived', 'completed'])
|
|
def test_historical_warning_preserves_source_and_allows_allocation(tmp_path, status):
|
|
archive, live = _sources(tmp_path, status=status)
|
|
before = {p: p.read_bytes() for p in (archive, live)}
|
|
preview = ensure_missing_work_record_identifiers(tmp_path, execute=False)
|
|
assert all(p.read_bytes() == content for p, content in before.items())
|
|
result = ensure_missing_work_record_identifiers(tmp_path)
|
|
assert result['assignments'] == preview['assignments']
|
|
assert {x['record_id'] for x in result['assignments']} == {'NK-WP-0099', 'NK-WP-0099-T01'}
|
|
assert archive.read_bytes() == before[archive]
|
|
assert result['historical_invalid_identifiers'] == scan_record_identities(tmp_path)['invalid_identifiers']
|
|
assert result['historical_invalid_identifiers'][0]['id'] == 'NK-WP-0003-T08a'
|
|
assert not scan_record_identities(tmp_path)['ok']
|
|
assert ensure_missing_work_record_identifiers(tmp_path)['assignments'] == []
|
|
|
|
|
|
@pytest.mark.parametrize('kwargs', [
|
|
{'status': 'active'}, {'status': 'unknown'}, {'task_status': 'todo'},
|
|
{'binding': ''}, {'binding': 'not-a-uuid'}, {'task_id': 'T08a'},
|
|
{'task_id': 'T01'}, {'task_id': 'NK-WP-0099-T08a'},
|
|
])
|
|
def test_unsafe_history_refuses_before_any_allocation(tmp_path, kwargs):
|
|
paths = _sources(tmp_path, **kwargs)
|
|
before = {p: p.read_bytes() for p in paths}
|
|
with pytest.raises(ValueError, match='identities are not safe'):
|
|
ensure_missing_work_record_identifiers(tmp_path)
|
|
assert all(p.read_bytes() == content for p, content in before.items())
|
|
|
|
|
|
def test_historical_collision_still_blocks(tmp_path):
|
|
archive, live = _sources(tmp_path)
|
|
duplicate = archive.with_name('duplicate.md')
|
|
duplicate.write_text(archive.read_text().replace(
|
|
'22222222-2222-4222-8222-222222222222',
|
|
'33333333-3333-4333-8333-333333333333'))
|
|
before = live.read_bytes()
|
|
with pytest.raises(ValueError, match='collisions='):
|
|
ensure_missing_work_record_identifiers(tmp_path)
|
|
assert live.read_bytes() == before
|