222 lines
7.4 KiB
Python
222 lines
7.4 KiB
Python
"""
|
|
Phase A — WebDAV Container Tests
|
|
|
|
Integration tests against a running obsidian-webdav container.
|
|
Skip these in CI unless the WebDAV service is available.
|
|
|
|
Run with:
|
|
pytest tests/test_phase_a_webdav.py -m integration -v
|
|
|
|
Environment variables required:
|
|
WEBDAV_URL e.g. http://100.x.x.x (Tailscale IP)
|
|
WEBDAV_USER basic-auth username (default: obsidian)
|
|
WEBDAV_PASS basic-auth password
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def webdav_base() -> str:
|
|
url = os.environ.get("WEBDAV_URL", "").rstrip("/")
|
|
if not url:
|
|
pytest.skip("WEBDAV_URL not set — skipping WebDAV integration tests")
|
|
return url
|
|
|
|
|
|
def webdav_auth():
|
|
return (
|
|
os.environ.get("WEBDAV_USER", "obsidian"),
|
|
os.environ.get("WEBDAV_PASS", ""),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def dav():
|
|
"""Requests session pre-configured for the WebDAV endpoint."""
|
|
s = requests.Session()
|
|
s.auth = webdav_auth()
|
|
return s
|
|
|
|
|
|
@pytest.fixture
|
|
def test_filename():
|
|
"""Unique filename so parallel test runs do not collide."""
|
|
return f"test_{uuid.uuid4().hex[:8]}.md"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-A2 — Authentication
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWebDAVAuth:
|
|
|
|
def test_unauthenticated_request_returns_401(self):
|
|
"""GET without credentials must be rejected."""
|
|
r = requests.get(webdav_base())
|
|
assert r.status_code == 401, f"Expected 401, got {r.status_code}"
|
|
|
|
def test_wrong_password_returns_401(self):
|
|
r = requests.get(webdav_base(), auth=("obsidian", "wrong-password"))
|
|
assert r.status_code == 401
|
|
|
|
def test_valid_credentials_succeed(self, dav):
|
|
r = dav.request("PROPFIND", webdav_base(), headers={"Depth": "0"})
|
|
assert r.status_code in (200, 207), f"Expected 200/207, got {r.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-A4 — WebDAV method support (PROPFIND, PUT, GET, DELETE)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWebDAVMethods:
|
|
|
|
def test_propfind_root_lists_contents(self, dav):
|
|
"""PROPFIND depth=1 must enumerate the root of the vault."""
|
|
r = dav.request(
|
|
"PROPFIND",
|
|
webdav_base(),
|
|
headers={"Depth": "1", "Content-Type": "application/xml"},
|
|
)
|
|
assert r.status_code in (200, 207)
|
|
assert len(r.content) > 0
|
|
|
|
def test_put_creates_file(self, dav, test_filename):
|
|
url = f"{webdav_base()}/{test_filename}"
|
|
content = b"# Test Note\nCreated by automated test.\n"
|
|
r = dav.put(url, data=content)
|
|
assert r.status_code in (200, 201, 204), f"PUT failed: {r.status_code}"
|
|
|
|
# Verify it exists
|
|
r2 = dav.get(url)
|
|
assert r2.status_code == 200
|
|
assert b"# Test Note" in r2.content
|
|
|
|
# Cleanup
|
|
dav.delete(url)
|
|
|
|
def test_put_updates_existing_file(self, dav, test_filename):
|
|
url = f"{webdav_base()}/{test_filename}"
|
|
dav.put(url, data=b"v1 content")
|
|
|
|
dav.put(url, data=b"v2 content updated")
|
|
r = dav.get(url)
|
|
assert b"v2 content" in r.content
|
|
|
|
dav.delete(url)
|
|
|
|
def test_delete_removes_file(self, dav, test_filename):
|
|
url = f"{webdav_base()}/{test_filename}"
|
|
dav.put(url, data=b"temporary file")
|
|
|
|
r = dav.delete(url)
|
|
assert r.status_code in (200, 204)
|
|
|
|
r2 = dav.get(url)
|
|
assert r2.status_code == 404
|
|
|
|
def test_get_nonexistent_returns_404(self, dav):
|
|
r = dav.get(f"{webdav_base()}/does_not_exist_{uuid.uuid4().hex}.md")
|
|
assert r.status_code == 404
|
|
|
|
def test_mkcol_creates_subdirectory(self, dav):
|
|
dirname = f"test_dir_{uuid.uuid4().hex[:8]}"
|
|
r = dav.request("MKCOL", f"{webdav_base()}/{dirname}")
|
|
assert r.status_code in (200, 201, 207)
|
|
|
|
r2 = dav.request("PROPFIND", f"{webdav_base()}/{dirname}", headers={"Depth": "0"})
|
|
assert r2.status_code in (200, 207)
|
|
|
|
dav.delete(f"{webdav_base()}/{dirname}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-A3 — .git/ directory protection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWebDAVGitProtection:
|
|
|
|
def test_git_directory_is_inaccessible(self, dav):
|
|
"""The .git/ directory must not be served — nginx should deny it."""
|
|
r = dav.request("PROPFIND", f"{webdav_base()}/.git/", headers={"Depth": "0"})
|
|
assert r.status_code in (403, 404), (
|
|
f"Expected 403/404 for /.git/ but got {r.status_code}. "
|
|
"The WebDAV config must deny access to .git/"
|
|
)
|
|
|
|
def test_git_config_file_is_inaccessible(self, dav):
|
|
r = dav.get(f"{webdav_base()}/.git/config")
|
|
assert r.status_code in (403, 404)
|
|
|
|
def test_git_head_file_is_inaccessible(self, dav):
|
|
r = dav.get(f"{webdav_base()}/.git/HEAD")
|
|
assert r.status_code in (403, 404)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-A1 — Bidirectional sync simulation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWebDAVFileOps:
|
|
|
|
def test_create_read_delete_roundtrip(self, dav, test_filename):
|
|
"""Full create → read → delete cycle for a markdown file."""
|
|
url = f"{webdav_base()}/{test_filename}"
|
|
body = "---\noutline_id: test-001\n---\n# Test\nContent.\n"
|
|
|
|
# Create
|
|
assert dav.put(url, data=body.encode()).status_code in (200, 201, 204)
|
|
|
|
# Read back — content must match
|
|
r = dav.get(url)
|
|
assert r.status_code == 200
|
|
assert "outline_id: test-001" in r.text
|
|
|
|
# Delete
|
|
assert dav.delete(url).status_code in (200, 204)
|
|
|
|
def test_unicode_content_preserved(self, dav, test_filename):
|
|
url = f"{webdav_base()}/{test_filename}"
|
|
body = "# Ünïcödé Héadïng\n\nGerman: Straße, Chinese: 你好\n".encode("utf-8")
|
|
|
|
dav.put(url, data=body)
|
|
r = dav.get(url)
|
|
assert "Straße" in r.text
|
|
assert "你好" in r.text
|
|
|
|
dav.delete(url)
|
|
|
|
def test_large_file_survives_roundtrip(self, dav, test_filename):
|
|
url = f"{webdav_base()}/{test_filename}"
|
|
# 500 KB markdown file
|
|
body = ("# Big Note\n" + "x" * 500_000).encode()
|
|
|
|
dav.put(url, data=body)
|
|
r = dav.get(url)
|
|
assert len(r.content) >= 500_000
|
|
|
|
dav.delete(url)
|
|
|
|
def test_obsidian_settings_directory_can_be_excluded(self, dav):
|
|
"""
|
|
Verify we can PUT to a path we'd want to ignore (.obsidian/)
|
|
and then verify the nginx/webdav config (if configured) can block it.
|
|
This test documents expected behaviour; if .obsidian/ IS accessible,
|
|
it must be controlled at the Obsidian plugin level (ignore list).
|
|
"""
|
|
# This is an informational check, not a hard assertion —
|
|
# .obsidian/ exclusion is handled by the remotely-save plugin config.
|
|
r = dav.request("PROPFIND", f"{webdav_base()}/.obsidian/", headers={"Depth": "0"})
|
|
# 404 = not present (preferred), 403 = blocked, 207 = accessible
|
|
# All are valid — important thing is it is documented
|
|
assert r.status_code in (403, 404, 207)
|