74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
from pathlib import Path
|
|
|
|
import httpx
|
|
from imagefind.sources import LocalConnector, WebDavConnector
|
|
|
|
|
|
def test_local_connector_ignores_symlinks_and_non_video(tmp_path: Path):
|
|
(tmp_path / "a.mp4").write_bytes(b"video")
|
|
(tmp_path / "note.txt").write_text("no")
|
|
outside = tmp_path.parent / "outside.mkv"
|
|
outside.write_bytes(b"outside")
|
|
(tmp_path / "link.mkv").symlink_to(outside)
|
|
items = list(LocalConnector(str(tmp_path)).items())
|
|
assert [item.key for item in items] == ["a.mp4"]
|
|
|
|
|
|
def test_webdav_depth_one_and_external_href_rejection():
|
|
xml = b"""<?xml version="1.0"?>
|
|
<d:multistatus xmlns:d="DAV:">
|
|
<d:response>
|
|
<d:href>/videos/</d:href><d:propstat><d:prop>
|
|
<d:resourcetype><d:collection/></d:resourcetype>
|
|
</d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat>
|
|
</d:response>
|
|
<d:response>
|
|
<d:href>/videos/movie.mp4</d:href><d:propstat><d:prop>
|
|
<d:resourcetype/><d:getcontentlength>123</d:getcontentlength><d:getetag>abc</d:getetag>
|
|
</d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat>
|
|
</d:response>
|
|
<d:response>
|
|
<d:href>https://evil.example/leak.mp4</d:href><d:propstat><d:prop>
|
|
<d:resourcetype/>
|
|
</d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat>
|
|
</d:response>
|
|
</d:multistatus>"""
|
|
|
|
def handler(request: httpx.Request):
|
|
assert request.method == "PROPFIND"
|
|
assert request.headers["Depth"] == "1"
|
|
return httpx.Response(207, content=xml)
|
|
|
|
connector = WebDavConnector("https://dav.example/videos/", "user", "pass")
|
|
connector.client.close()
|
|
connector.client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
items = list(connector.items())
|
|
connector.close()
|
|
assert len(items) == 1
|
|
assert items[0].key == "movie.mp4"
|
|
assert items[0].etag == "abc"
|
|
|
|
|
|
def test_webdav_tolerates_alist_xml_with_extra_payload():
|
|
xml = b"""alist notice
|
|
<D:multistatus xmlns:D="DAV:">
|
|
<D:response>
|
|
<D:href>/dav/movie.mp4</D:href><D:propstat><D:prop>
|
|
<D:resourcetype/><D:getcontentlength>42</D:getcontentlength>
|
|
</D:prop><D:status>HTTP/1.1 200 OK</D:status></D:propstat>
|
|
</D:response>
|
|
</D:multistatus><ignored/>"""
|
|
|
|
def handler(_request: httpx.Request):
|
|
return httpx.Response(207, content=xml)
|
|
|
|
connector = WebDavConnector("https://dav.example/dav/", "user", "pass")
|
|
connector.client.close()
|
|
connector.client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
items = list(connector.items())
|
|
connector.close()
|
|
|
|
assert len(items) == 1
|
|
assert items[0].key == "movie.mp4"
|
|
assert items[0].size_bytes == 42
|