108 lines
3.4 KiB
Python
108 lines
3.4 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'], # noqa S607 -- test run in a controlled environment
|
|
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'
|
|
|
|
entry_count = 4
|
|
for index in range(entry_count):
|
|
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 `entry_count` entries
|
|
backend.flush()
|
|
log_lines = _get_audit_log_lines(backend, record_id)
|
|
assert len(log_lines) == entry_count
|
|
|
|
# Check that the commit log has `entry_count` + 1 (from `README.txt`) entries
|
|
commit_log_lines = _get_git_log(tmp_path)
|
|
assert len(commit_log_lines) == entry_count + 1
|
|
|
|
# Check that the changes are reported
|
|
changes = backend.get_audit_log(record_id)
|
|
assert len(changes) == entry_count
|
|
assert tuple(e[0:2] for e in 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',
|
|
)
|
|
|
|
expected_entries = 1
|
|
|
|
# Check that there is only one entry in the audit log
|
|
log_lines = _get_audit_log_lines(backend, record_id)
|
|
assert len(log_lines) == expected_entries
|
|
|
|
# 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) == expected_entries + 1
|
|
|
|
# Check that the changes are reported
|
|
changes = backend.get_audit_log(record_id)
|
|
assert len(changes) == expected_entries
|
|
|
|
|
|
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
|