sashatrask
113 lines
4.6 KiB
Python
113 lines
4.6 KiB
Python
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from test_docker_codec_recovery import recovery
|
|
from test_docker_layer_scanning import docker_plan, layer_work
|
|
import scanner
|
|
|
|
|
|
@pytest.fixture(params=['layer', 'full'])
|
|
def producer(request, recovery):
|
|
if request.param == 'full':
|
|
return SimpleNamespace(run=recovery.run, blob_count=4)
|
|
layer = dict(recovery.resolved['layers'][1], position=1, selected=True,
|
|
selection_reason='layer_selected', coverage_state='leased',
|
|
lease_token='k' * 43, attempt=1, max_attempts=3)
|
|
plan = docker_plan(b'{}', extra_descriptors=(layer,))
|
|
recovery.payloads[plan['descriptors'][0]['digest']] = b'{}'
|
|
work = layer_work(plan)
|
|
return SimpleNamespace(run=lambda: scanner.scan_docker_layer_plan(plan['image'], work), blob_count=2)
|
|
|
|
|
|
def test_fatal_scan_retains_payload_and_root_without_masking_fatal(producer, monkeypatch):
|
|
fatal = scanner.ScanSlotFatalError('synthetic child termination is unconfirmed')
|
|
payloads = []
|
|
|
|
def scan(path, *args, **kwargs):
|
|
payloads.append(Path(path))
|
|
raise fatal
|
|
|
|
unlink = mock.Mock(side_effect=PermissionError('synthetic Windows sharing violation'))
|
|
cleanup = mock.Mock(side_effect=RuntimeError('must not clean an unconfirmed tree'))
|
|
monkeypatch.setattr(scanner, '_scan_docker_content_file', scan)
|
|
monkeypatch.setattr(scanner, 'durable_unlink', unlink)
|
|
monkeypatch.setattr(scanner, 'cleanup_command_work_dir', cleanup)
|
|
with pytest.raises(scanner.ScanSlotFatalError) as raised:
|
|
producer.run()
|
|
assert raised.value is fatal
|
|
unlink.assert_not_called()
|
|
cleanup.assert_not_called()
|
|
assert len(payloads) == 1 and payloads[0].is_file()
|
|
assert payloads[0].parent.is_dir() and payloads[0].parent.name.startswith('docker-layer-')
|
|
|
|
|
|
def test_fatal_in_unlink_is_not_retried_or_masked_by_outer_cleanup(producer, monkeypatch):
|
|
fatal = scanner.ScanSlotFatalError('synthetic child guard rejected unlink')
|
|
payloads = []
|
|
|
|
def scan(path, *args, **kwargs):
|
|
payloads.append(Path(path))
|
|
return {'findings': [], 'errors': []}
|
|
|
|
unlink = mock.Mock(side_effect=fatal)
|
|
cleanup = mock.Mock(side_effect=PermissionError('must not mask the fatal error'))
|
|
monkeypatch.setattr(scanner, '_scan_docker_content_file', scan)
|
|
monkeypatch.setattr(scanner, 'durable_unlink', unlink)
|
|
monkeypatch.setattr(scanner, 'cleanup_command_work_dir', cleanup)
|
|
with pytest.raises(scanner.ScanSlotFatalError) as raised:
|
|
producer.run()
|
|
assert raised.value is fatal
|
|
unlink.assert_called_once()
|
|
cleanup.assert_not_called()
|
|
assert payloads[0].is_file() and payloads[0].parent.is_dir()
|
|
|
|
|
|
def test_later_fatal_retains_active_payload_but_completed_blob_was_cleaned(producer, monkeypatch):
|
|
fatal = scanner.ScanSlotFatalError('synthetic second child is unconfirmed')
|
|
seen = []
|
|
unlink = mock.Mock(wraps=scanner.durable_unlink)
|
|
cleanup = mock.Mock(side_effect=AssertionError('fatal root cleanup attempted'))
|
|
|
|
def scan(path, *args, **kwargs):
|
|
seen.append(Path(path))
|
|
if len(seen) == 2:
|
|
raise fatal
|
|
return {'findings': [], 'errors': []}
|
|
|
|
monkeypatch.setattr(scanner, '_scan_docker_content_file', scan)
|
|
monkeypatch.setattr(scanner, 'durable_unlink', unlink)
|
|
monkeypatch.setattr(scanner, 'cleanup_command_work_dir', cleanup)
|
|
with pytest.raises(scanner.ScanSlotFatalError) as raised:
|
|
producer.run()
|
|
assert raised.value is fatal
|
|
unlink.assert_called_once_with(str(seen[0]))
|
|
cleanup.assert_not_called()
|
|
assert seen[1].is_file() and seen[1].parent.is_dir()
|
|
|
|
|
|
@pytest.mark.parametrize('scan_error', [False, True])
|
|
def test_normal_and_nonfatal_scan_cleanup_remain_enabled(producer, monkeypatch, scan_error):
|
|
seen = []
|
|
unlink = mock.Mock(wraps=scanner.durable_unlink)
|
|
cleanup = mock.Mock(wraps=scanner.cleanup_command_work_dir)
|
|
|
|
def scan(path, *args, **kwargs):
|
|
seen.append(Path(path))
|
|
if scan_error:
|
|
raise scanner.DockerContentScanError('invalid_layer_archive', 'synthetic malformed archive')
|
|
return {'findings': [], 'errors': []}
|
|
|
|
monkeypatch.setattr(scanner, '_scan_docker_content_file', scan)
|
|
monkeypatch.setattr(scanner, 'durable_unlink', unlink)
|
|
monkeypatch.setattr(scanner, 'cleanup_command_work_dir', cleanup)
|
|
result = producer.run()
|
|
assert bool(result['errors']) == scan_error
|
|
assert unlink.call_count == len(seen)
|
|
if not scan_error:
|
|
assert len(seen) == producer.blob_count
|
|
cleanup.assert_called_once()
|
|
assert all(not path.exists() and not path.parent.exists() for path in seen)
|