tegwick
cb60a00ee1
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
102 lines
4.5 KiB
Python
102 lines
4.5 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
|
|
|
|
|
|
def _railiance_item(root: Path, *, parent="RAILIANCE-WP-0004", status="finished",
|
|
task_status="done", task_id="RAILIANCE-WP-0004-I01",
|
|
binding="44444444-4444-4444-8444-444444444444") -> Path:
|
|
path = root / "workplans" / "archived" / "railiance.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
f'---\nid: {parent}\nstatus: {status}\n'
|
|
'state_hub_workstream_id: "33333333-3333-4333-8333-333333333333"\n---\n'
|
|
f'\n```task\nid: {task_id}\nstatus: {task_status}\n'
|
|
f'state_hub_task_id: "{binding}"\n```\n',
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_railiance_historical_item_labels_preserve_uuid(tmp_path: Path) -> None:
|
|
path = _railiance_item(tmp_path)
|
|
before = path.read_bytes()
|
|
result = ensure_missing_work_record_identifiers(tmp_path, execute=False)
|
|
assert result["assignments"] == []
|
|
assert result["files_changed"] == []
|
|
assert path.read_bytes() == before
|
|
|
|
|
|
@pytest.mark.parametrize("kwargs", [
|
|
{"status": "active"},
|
|
{"task_status": "todo"},
|
|
{"parent": "RAILIANCE-WP-0005", "task_id": "RAILIANCE-WP-0005-I01"},
|
|
{"binding": ""},
|
|
{"binding": "not-a-uuid"},
|
|
{"task_id": "RAILIANCE-WP-0004-I08"},
|
|
{"task_id": "OTHER-WP-0004-I01"},
|
|
])
|
|
def test_railiance_item_label_preservation_is_bounded(tmp_path: Path, kwargs) -> None:
|
|
_railiance_item(tmp_path, **kwargs)
|
|
with pytest.raises(ValueError, match="identities are not safe"):
|
|
ensure_missing_work_record_identifiers(tmp_path)
|