55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.modules.data_process.store import (
|
||
|
|
DataProcessStoreError,
|
||
|
|
_source_storage_descriptor,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_source_storage_descriptor_accepts_owned_local_and_legacy_db_references() -> None:
|
||
|
|
task_id = "dpt_task"
|
||
|
|
source_file_id = "dpsf_source"
|
||
|
|
local_reference = (
|
||
|
|
f"local://data-process/{task_id}/{source_file_id}/v1/source%20100%25.csv"
|
||
|
|
)
|
||
|
|
|
||
|
|
reference, metadata = _source_storage_descriptor(
|
||
|
|
{
|
||
|
|
"storage_object_id": local_reference,
|
||
|
|
"metadata": {"storage_backend": "spoofed", "content_type": "text/csv"},
|
||
|
|
},
|
||
|
|
task_id,
|
||
|
|
source_file_id,
|
||
|
|
)
|
||
|
|
assert reference == local_reference
|
||
|
|
assert metadata == {"storage_backend": "local", "content_type": "text/csv"}
|
||
|
|
|
||
|
|
legacy_reference, legacy_metadata = _source_storage_descriptor(
|
||
|
|
{"metadata": {"legacy": True}},
|
||
|
|
task_id,
|
||
|
|
source_file_id,
|
||
|
|
)
|
||
|
|
assert legacy_reference == f"db://data-process/{task_id}/{source_file_id}/v1"
|
||
|
|
assert legacy_metadata == {"storage_backend": "database", "legacy": True}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"reference",
|
||
|
|
[
|
||
|
|
"local://data-process/dpt_other/dpsf_source/v1/source.txt",
|
||
|
|
"db://data-process/dpt_task/dpsf_other/v1",
|
||
|
|
"/var/tmp/source.txt",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
def test_source_storage_descriptor_rejects_unowned_or_unsupported_references(
|
||
|
|
reference: str,
|
||
|
|
) -> None:
|
||
|
|
with pytest.raises(DataProcessStoreError):
|
||
|
|
_source_storage_descriptor(
|
||
|
|
{"storage_object_id": reference},
|
||
|
|
"dpt_task",
|
||
|
|
"dpsf_source",
|
||
|
|
)
|