commit 9143a54ab31be446c5ee8daec260a11109c1d2a8
parent be0e8b075b2f40dad5c14d5b30c9109ddc85d8cb
Author: archiveanon <>
Date: Sun, 7 Sep 2025 20:43:40 +0000
Add WebDAV upload test
Diffstat:
3 files changed, 79 insertions(+), 0 deletions(-)
diff --git a/pyproject.toml b/pyproject.toml
@@ -26,6 +26,8 @@ autotako = "autotako.app:main"
dev = [
"mypy == 1.9.0",
"ruff == 0.3.7",
+ "pytest",
+ "pytest-asyncio",
]
[build-system]
@@ -39,3 +41,9 @@ line-length = 96
[tool.setuptools.package-data]
autotako = [ "static/*", "templates/*" ]
+
+[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "function"
+markers = [
+ "webdav: tests for integration with the webdav service",
+]
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -0,0 +1,40 @@
+#!/usr/bin/python3
+
+import pathlib
+
+import msgspec
+import pytest
+from autotako.config import WebDavConfig
+
+
+class WebDavTestConfig(msgspec.Struct):
+ webdav: WebDavConfig
+ path: str = "test/upload.bin"
+ """ Path to upload a test file to, including filename. """
+
+
+class Config(msgspec.Struct):
+ test_webdav: WebDavTestConfig
+
+
+def pytest_addoption(parser):
+ parser.addoption(
+ "--test-config",
+ action="store",
+ type=pathlib.Path,
+ default="test_config.toml",
+ help="Path to test configuration file",
+ )
+
+
+@pytest.fixture
+def config(request, scope="session"):
+ return msgspec.toml.decode(
+ request.config.getoption("--test-config").read_text(),
+ type=Config,
+ )
+
+
+@pytest.fixture
+def webdav_test_config(config):
+ return config.test_webdav
diff --git a/tests/test_webdav_upload.py b/tests/test_webdav_upload.py
@@ -0,0 +1,31 @@
+#!/usr/bin/python3
+
+import secrets
+import urllib
+
+import httpx
+import pytest
+from autotako.job_render import do_webdav_upload
+
+
+@pytest.mark.asyncio
+@pytest.mark.webdav
+async def test_webdav_upload(webdav_test_config, tmp_path):
+ webdav = webdav_test_config.webdav
+ test_path = webdav_test_config.path
+
+ buffer = secrets.token_bytes(4096)
+
+ auth = httpx.BasicAuth(username=webdav.username, password=webdav.password)
+ async with httpx.AsyncClient(auth=auth) as webclient:
+ await webclient.delete(urllib.parse.urljoin(webdav.base_url, test_path))
+
+ tf = tmp_path / "test.bin"
+ tf.write_bytes(buffer)
+ await do_webdav_upload(webdav, tf, test_path)
+
+ resp = await webclient.get(urllib.parse.urljoin(webdav.base_url, test_path))
+
+ assert buffer == resp.content
+
+ await webclient.delete(urllib.parse.urljoin(webdav.base_url, test_path))