29 lines
1.1 KiB
Python
29 lines
1.1 KiB
Python
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from ingestor.connection.certs import CertProvider
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_returns_local_paths(tmp_path):
|
|
mongo = AsyncMock()
|
|
mongo.find_one.return_value = {
|
|
'client_cert': {'minio_object_name': 'platform/certificates/client_cert_c.pem'},
|
|
'private_key': {'minio_object_name': 'platform/certificates/private_key_k.pem'},
|
|
}
|
|
minio = AsyncMock()
|
|
minio.download_file.side_effect = [b'cert-bytes', b'key-bytes']
|
|
|
|
provider = CertProvider(mongo, minio, cache_dir=str(tmp_path), bucket='sientia-dataops')
|
|
cert, key = await provider.fetch()
|
|
|
|
assert cert.endswith('c.pem') and key.endswith('k.pem')
|
|
assert minio.download_file.await_count == 2
|
|
mongo.find_one.assert_awaited_once_with('platform_certificates', {}, {})
|
|
minio.download_file.assert_any_await(
|
|
'platform/certificates/client_cert_c.pem', 'sientia-dataops', {}
|
|
)
|
|
assert (tmp_path / 'client_cert_c.pem').read_bytes() == b'cert-bytes'
|
|
assert (tmp_path / 'private_key_k.pem').read_bytes() == b'key-bytes'
|