152 lines
5.2 KiB
Python
152 lines
5.2 KiB
Python
import json
|
|
import stat
|
|
|
|
import pytest
|
|
|
|
from pigway_frame_flow.errors import FlowError
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
from pigway_frame_flow.providers.synology_photos import PhotosLogin, SynologyPhotos
|
|
|
|
|
|
def test_connection_stores_session_only(settings, monkeypatch):
|
|
p = SynologyPhotos(settings)
|
|
monkeypatch.setenv("SYNOLOGY_PHOTOS_URL", "https://nas.test:9115")
|
|
monkeypatch.setenv("SYNOLOGY_PHOTOS_CERT_SHA256", "a" * 64)
|
|
monkeypatch.setattr(p, "request", lambda *args, **kwargs: {"sid": "test-session"})
|
|
assert p.connect(PhotosLogin(username="way", password="test-password"))["connected"]
|
|
assert json.loads(p.path.read_text()) == {"sid": "test-session"}
|
|
assert stat.S_IMODE(p.path.stat().st_mode) == 0o600
|
|
assert "test-password" not in p.path.read_text()
|
|
assert p.sid() == "test-session"
|
|
assert (
|
|
settings.sanitize({"message": "test-password test-session"})["message"]
|
|
== "[REDACTED] [REDACTED]"
|
|
)
|
|
assert not p.disconnect()["connected"]
|
|
|
|
|
|
def test_certificate_checked_before_password(settings, monkeypatch):
|
|
monkeypatch.setenv("SYNOLOGY_PHOTOS_URL", "https://nas.test:9115")
|
|
monkeypatch.setenv("SYNOLOGY_PHOTOS_CERT_SHA256", "a" * 64)
|
|
sent = []
|
|
|
|
class Socket:
|
|
def getpeercert(self, **kwargs):
|
|
return b"wrong-cert"
|
|
|
|
class Connection:
|
|
sock = Socket()
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def connect(self):
|
|
pass
|
|
|
|
def request(self, *args):
|
|
sent.append(args)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
monkeypatch.setattr("http.client.HTTPSConnection", Connection)
|
|
with pytest.raises(FlowError) as e:
|
|
SynologyPhotos(settings).connect(
|
|
PhotosLogin(username="way", password="test-password")
|
|
)
|
|
assert e.value.status == 403 and sent == []
|
|
|
|
|
|
def test_mutating_rpc_blocked(settings):
|
|
with pytest.raises(FlowError) as e:
|
|
SynologyPhotos(settings).request("SYNO.FotoTeam.Browse.Item", "delete", 7)
|
|
assert e.value.status == 403
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mode,expected",
|
|
[
|
|
("any", {"20240101 Day/a.jpg", "20240101 Day/b.jpg"}),
|
|
("all", {"20240101 Day/b.jpg"}),
|
|
],
|
|
)
|
|
def test_person_paths_intersection_and_cache(settings, monkeypatch, mode, expected):
|
|
p = SynologyPhotos(settings)
|
|
monkeypatch.setattr(p, "people", lambda: [{"id": 1}, {"id": 2}])
|
|
calls = []
|
|
|
|
def pages(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
names = ["a.jpg", "b.jpg"] if kwargs["person_id"] == 1 else ["b.jpg"]
|
|
return [
|
|
{"filename": n, "additional": {"folder": "/20240101 Day"}} for n in names
|
|
]
|
|
|
|
monkeypatch.setattr(p, "pages", pages)
|
|
assert p.paths([1, 2], mode) == expected
|
|
assert p.paths([2, 1], mode) == expected
|
|
assert len(calls) == 2
|
|
with pytest.raises(FlowError):
|
|
p.paths([999], mode)
|
|
|
|
|
|
def test_missing_folder_fails_closed(settings, monkeypatch):
|
|
p = SynologyPhotos(settings)
|
|
monkeypatch.setattr(p, "people", lambda: [{"id": 1}])
|
|
monkeypatch.setattr(p, "pages", lambda *a, **k: [{"filename": "a.jpg"}])
|
|
with pytest.raises(FlowError):
|
|
p.paths([1], "any")
|
|
|
|
|
|
def test_source_people_filter_and_no_fallback(service, settings, monkeypatch):
|
|
from PIL import Image
|
|
|
|
settings.photo_root = settings.data_dir.parent / "photos"
|
|
for name in ["20240101 Day/a.jpg", "20240101 Day/b.jpg", "backup/c.jpg"]:
|
|
path = settings.photo_root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (10, 10)).save(path)
|
|
source = PhotoSourceInput(
|
|
name="People",
|
|
kind="local",
|
|
location=".",
|
|
organized_only=True,
|
|
folder_caption=True,
|
|
people_filter=True,
|
|
person_ids=[1],
|
|
)
|
|
p = service.photo_sources
|
|
p.save("people", source)
|
|
monkeypatch.setattr(
|
|
p.synology, "paths", lambda *a, **k: {"20240101 Day/b.jpg", "backup/c.jpg"}
|
|
)
|
|
rows = p.catalog("people")
|
|
assert len(rows) == 1 and rows[0]["id"] == "20240101 Day/b.jpg"
|
|
assert rows[0]["caption"]["date"] == "2024-01-01"
|
|
monkeypatch.setattr(p.synology, "paths", lambda *a, **k: set())
|
|
assert p.catalog("people", refresh=True) == []
|
|
with pytest.raises(FlowError):
|
|
p.read("people")
|
|
with pytest.raises(FlowError):
|
|
p.save("empty", source.model_copy(update={"person_ids": []}))
|
|
with pytest.raises(FlowError):
|
|
p.save("dav", source.model_copy(update={"kind": "webdav"}))
|
|
|
|
|
|
def test_people_api_auth_and_secret_response(client, service, monkeypatch):
|
|
p = service.photo_sources.synology
|
|
monkeypatch.setattr(
|
|
p, "connect", lambda body: {"configured": True, "connected": True}
|
|
)
|
|
r = client.post(
|
|
"/api/v1/integrations/synology-photos",
|
|
json={"username": "way", "password": "test-password"},
|
|
)
|
|
assert r.status_code == 200 and "password" not in r.text
|
|
r = client.post(
|
|
"/api/v1/integrations/synology-photos", json={"username": "way", "password": ""}
|
|
)
|
|
assert r.status_code == 422 and "input" not in r.text
|
|
client.headers.pop("Authorization")
|
|
assert client.get("/api/v1/integrations/synology-photos/people").status_code == 401
|