pigway-frame-flow/tests/test_synology_photos.py

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({"name": "pigway-frame-flow"})["name"] == "pigway-frame-flow"
    )
    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