2d9e0c405d
Some checks failed
Test execution / Test-all (push) Failing after 42s
Use the code from the minilad branch of datalad-core.
105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
import subprocess
|
|
from pathlib import Path
|
|
|
|
from graphql.language import location
|
|
|
|
from dump_things_service.audit.gitaudit import GitAuditBackend
|
|
|
|
|
|
def _get_git_log(path: Path) -> list[str]:
|
|
result = subprocess.run(
|
|
['git', '-C', str(path), 'log', '--oneline'],
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
return result.stdout.decode().splitlines()
|
|
|
|
|
|
def _get_audit_log_lines(backend: GitAuditBackend, record_id: str) -> list[str]:
|
|
locations = backend.get_location_for(record_id)
|
|
return backend.read_from_repo_path(locations[2]).decode().splitlines()
|
|
|
|
|
|
def test_gitaudit_basic(tmp_path_factory):
|
|
tmp_path = tmp_path_factory.mktemp("gitaudit_backend")
|
|
|
|
backend = GitAuditBackend(tmp_path)
|
|
|
|
record_id = 'test_gitaudit_basic'
|
|
|
|
for index in range(4):
|
|
backend.add_record(
|
|
record_id=record_id,
|
|
record={'pid': record_id, 'content': index},
|
|
user_id=f'tester_{index}@example.com',
|
|
)
|
|
|
|
# Check that the log file has 4 entries
|
|
backend.flush()
|
|
log_lines = _get_audit_log_lines(backend, record_id)
|
|
assert len(log_lines) == 4
|
|
|
|
# Check that the commit log has 4 + 1 (from `README.txt`) entries
|
|
commit_log_lines = _get_git_log(tmp_path)
|
|
assert len(commit_log_lines) == 5
|
|
|
|
# Check that the changes are reported
|
|
changes = backend.get_audit_log(record_id)
|
|
assert len(changes) == 4
|
|
assert tuple(map(lambda e: e[0], changes.values())) == tuple((f'tester_{i}@example.com' for i in range(4)))
|
|
|
|
|
|
def test_gitaudit_identical_change(tmp_path_factory):
|
|
tmp_path = tmp_path_factory.mktemp("gitaudit_backend")
|
|
|
|
backend = GitAuditBackend(tmp_path)
|
|
|
|
record_id = 'test_gitaudit_idempotent'
|
|
backend.add_record(
|
|
record_id=record_id,
|
|
record={'pid': record_id},
|
|
user_id='tester@example.com',
|
|
)
|
|
|
|
backend.add_record(
|
|
record_id=record_id,
|
|
record={'pid': record_id},
|
|
user_id='tester@example.com',
|
|
)
|
|
|
|
# Check that there is only one entry in the audit log
|
|
log_lines = _get_audit_log_lines(backend, record_id)
|
|
assert len(log_lines) == 1
|
|
|
|
# Check that there are two entries in the commit history, one for the
|
|
# `README.txt`-file, one for the log entries.
|
|
commit_log_lines = _get_git_log(tmp_path)
|
|
assert len(commit_log_lines) == 2
|
|
|
|
# Check that the changes are reported
|
|
changes = backend.get_audit_log(record_id)
|
|
assert len(changes) == 1
|
|
|
|
|
|
def test_gitaudit_huge_log(tmp_path_factory):
|
|
tmp_path = tmp_path_factory.mktemp("gitaudit_backend")
|
|
|
|
backend = GitAuditBackend(tmp_path)
|
|
|
|
change_number = 2
|
|
record_number = 100
|
|
|
|
for j in range(change_number):
|
|
for i in range(record_number):
|
|
record_id = f'huge_{i}'
|
|
backend.add_record(
|
|
record_id=record_id,
|
|
record={'pid': record_id, 'content': f'i:{i}, j:{j}'},
|
|
user_id=f'tester_ij_{i}_{j}@example.com',
|
|
)
|
|
|
|
# Check that the changes are reported
|
|
for i in range(record_number):
|
|
record_id = f'huge_{i}'
|
|
changes = backend.get_audit_log(record_id)
|
|
assert len(changes) == change_number
|