c086fdaf56
All checks were successful
Test execution / Test-all (push) Successful in 2m4s
Record the committer ID and the author ID seperately in audit backends.
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
import subprocess
|
|
from pathlib import Path
|
|
|
|
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={'pid': record_id, 'content': index},
|
|
committer_id=f'committer_{100 + index}@x.org',
|
|
author_id=f'author_{index}@y.org',
|
|
)
|
|
|
|
# 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:2], changes.values())) == tuple(
|
|
(f'committer_{100 + i}@x.org', f'author_{i}@y.org')
|
|
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={'pid': record_id},
|
|
committer_id='committer_b@x.org',
|
|
author_id = 'author_b@y.org',
|
|
)
|
|
|
|
backend.add_record(
|
|
record={'pid': record_id},
|
|
committer_id='committer_b@x.org',
|
|
author_id = 'author_b@y.org',
|
|
)
|
|
|
|
# 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={'pid': record_id, 'content': f'j:{j}, i:{i}'},
|
|
committer_id='committer@x.org',
|
|
author_id = 'author@y.org',
|
|
)
|
|
|
|
# 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
|