96 lines
4.4 KiB
Python
96 lines
4.4 KiB
Python
import copy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from parser_lab.admin import config_from_payload
|
|
from parser_lab.repository import Repository, Conflict, NotFound
|
|
from parser_lab.views import default_view, project_record, validate_view
|
|
from parser_lab.web import public_page
|
|
|
|
|
|
def test_detail_fields_are_independent_and_old_views_keep_their_fields():
|
|
row = {'name': 'Alice', 'bio': 'Detailed biography', 'secret': 'PRIVATE'}
|
|
view = default_view('table', ['/name'])
|
|
view['detail_columns'] = [{'pointer': '/bio', 'label': 'Biography'}]
|
|
publication = {'title': 'Study', 'description': '', 'version': 1,
|
|
'records': [row], 'views': {'table': view}}
|
|
listing = public_page(publication, 'table', '/collection')
|
|
detail = public_page(publication, 'table', '/collection', record_id='1')
|
|
assert 'Detailed biography' not in listing
|
|
assert 'Detailed biography' in detail and 'Biography' in detail
|
|
assert 'PRIVATE' not in detail
|
|
assert [f['pointer'] for f in project_record(row, view, 1)['fields']] == ['/name']
|
|
del view['detail_columns']
|
|
legacy = public_page(publication, 'table', '/collection', record_id='1')
|
|
assert 'Alice' in legacy and 'Detailed biography' not in legacy
|
|
|
|
|
|
def test_detail_config_survives_save_and_validates_fields():
|
|
draft = {'revision': 1, 'records': [{'name': 'Alice', 'bio': 'About'}]}
|
|
view = default_view('table', ['/name'])
|
|
view['detail_columns'] = [{'pointer': '/bio', 'label': 'About'}]
|
|
result = config_from_payload(copy.deepcopy(draft), {'revision': 1, 'views': {'table': view}})
|
|
assert result['views']['table']['detail_columns'] == [{'pointer': '/bio', 'label': 'About'}]
|
|
view['detail_columns'] = [{'pointer': '/unknown', 'label': 'Unknown'}]
|
|
assert validate_view(view, draft['records'])
|
|
view['detail_columns'] = []
|
|
assert validate_view(view, draft['records'])
|
|
view['detail_columns'] = 'bad'
|
|
with pytest.raises(ValueError):
|
|
config_from_payload(copy.deepcopy(draft), {'revision': 1, 'views': {'table': view}})
|
|
|
|
|
|
def test_questionnaire_detail_fields_do_not_replace_answers():
|
|
row = {'name': 'Alice', 'bio': 'About Alice', 'answers': [{'question': 'Sleep?', 'answer': 7}]}
|
|
view = default_view('questionnaire', ['/name'])
|
|
view.update(questionnaire_mode='pairs', questions_pointer='/answers',
|
|
detail_columns=[{'pointer': '/bio', 'label': 'Bio'}])
|
|
fields = project_record(row, view, 1, detail=True)['fields']
|
|
assert [(f['label'], f['text']) for f in fields] == [('Bio', 'About Alice'), ('Sleep?', '7')]
|
|
|
|
|
|
def test_delete_removes_versions_and_sources_but_preserves_other_datasets(tmp_path):
|
|
repo = Repository(tmp_path)
|
|
d = repo.create('Delete me', b'old', 'old.json', [{'name': 'Alice'}])
|
|
other = repo.create('Keep me', b'keep', 'keep.json', [{'name': 'Bob'}])
|
|
d['views'] = {'table': default_view('table', ['/name'])}
|
|
d = repo.save(d, d['revision'])
|
|
repo.publish(d['id'], d['revision'])
|
|
repo.unpublish(d['slug'], 1)
|
|
for name in ('middle', 'latest'):
|
|
d = repo.replace_source(d['id'], d['revision'], name.encode(), name + '.json', [{'name': name}])
|
|
with pytest.raises(Conflict):
|
|
repo.delete(d['id'], d['revision'] - 1)
|
|
assert len(list((tmp_path / 'sources').iterdir())) == 4
|
|
repo.delete(d['id'], d['revision'])
|
|
with pytest.raises(NotFound):
|
|
repo.get(d['id'])
|
|
assert repo.versions(d['slug']) == []
|
|
assert repo.get(other['id'])['title'] == 'Keep me'
|
|
assert [p.read_bytes() for p in (tmp_path / 'sources').iterdir()] == [b'keep']
|
|
with pytest.raises(NotFound):
|
|
repo.delete(d['id'], d['revision'])
|
|
|
|
|
|
def test_delete_retries_failed_source_cleanup_on_restart(tmp_path, monkeypatch):
|
|
repo = Repository(tmp_path)
|
|
d = repo.create('Delete', b'old', 'old.json', [{'name': 'Alice'}])
|
|
d = repo.replace_source(d['id'], d['revision'], b'new', 'new.json', [{'name': 'Bob'}])
|
|
unlink = Path.unlink
|
|
|
|
def blocked_source(path, *args, **kwargs):
|
|
if path.name == d['source']:
|
|
raise PermissionError('File temporarily unavailable')
|
|
return unlink(path, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Path, 'unlink', blocked_source)
|
|
assert repo.delete(d['id'], d['revision']) is False
|
|
with pytest.raises(NotFound):
|
|
repo.get(d['id'])
|
|
assert (tmp_path / 'sources' / d['source']).read_bytes() == b'new'
|
|
monkeypatch.undo()
|
|
restarted = Repository(tmp_path)
|
|
assert restarted.list() == []
|
|
assert not list((tmp_path / 'sources').iterdir())
|