You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
63 lines
2.1 KiB
63 lines
2.1 KiB
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.adapters.storage.webdav import WebDAVStorageAdapter
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_webdav_adapter_supports_check_upload_download_and_exists(
|
|
tmp_path: Path,
|
|
fake_webdav: dict[str, object],
|
|
) -> None:
|
|
adapter = WebDAVStorageAdapter(
|
|
{
|
|
"endpoint_url": fake_webdav["endpoint_url"],
|
|
"username": fake_webdav["username"],
|
|
"password": fake_webdav["password"],
|
|
"root_path": "iron",
|
|
}
|
|
)
|
|
|
|
check_result = await adapter.check()
|
|
assert check_result["status"] == "healthy"
|
|
assert check_result["detail"] == "iron"
|
|
|
|
source = tmp_path / "sample.txt"
|
|
source.write_text("openlist-webdav", encoding="utf-8")
|
|
|
|
put_result = await adapter.put_file(str(source), "blobs/ab/sample.txt")
|
|
assert put_result["storage_key"] == "blobs/ab/sample.txt"
|
|
assert await adapter.object_exists("blobs/ab/sample.txt") is True
|
|
assert await adapter.object_exists("blobs/ab/missing.txt") is False
|
|
|
|
target = tmp_path / "downloaded.txt"
|
|
result_path = await adapter.download_file("blobs/ab/sample.txt", target)
|
|
assert result_path.read_text(encoding="utf-8") == "openlist-webdav"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_backend_route_can_check_webdav(api_client, fake_webdav: dict[str, object]) -> None:
|
|
create_response = await api_client.post(
|
|
"/api/backends",
|
|
json={
|
|
"name": "openlist-main",
|
|
"type": "webdav",
|
|
"stability_class": "stable",
|
|
"read_priority": 65,
|
|
"write_priority": 65,
|
|
"config": {
|
|
"endpoint_url": fake_webdav["endpoint_url"],
|
|
"username": fake_webdav["username"],
|
|
"password": fake_webdav["password"],
|
|
"root_path": "iron",
|
|
},
|
|
},
|
|
)
|
|
assert create_response.status_code == 200
|
|
|
|
backend_id = create_response.json()["backend"]["id"]
|
|
check_response = await api_client.post(f"/api/backends/{backend_id}/check")
|
|
assert check_response.status_code == 200
|
|
assert check_response.json()["status"] == "healthy"
|
|
assert check_response.json()["detail"] == "iron"
|
|
|