feat: initial Dropbox tag sync tool
CI / Tests (Linux, Python 3.12) (push) Failing after 4m51s
CI / Tests (Linux, Python 3.11) (push) Failing after 5m19s
CI / Tests (Linux, Python 3.9) (push) Failing after 5m3s
CI / Tests (macOS) (push) Has been cancelled
CI / Tests (Windows) (push) Has been cancelled
CI / Tests (Linux, Python 3.12) (push) Failing after 4m51s
CI / Tests (Linux, Python 3.11) (push) Failing after 5m19s
CI / Tests (Linux, Python 3.9) (push) Failing after 5m3s
CI / Tests (macOS) (push) Has been cancelled
CI / Tests (Windows) (push) Has been cancelled
Cross-platform CLI that mirrors Dropbox web tags to OS-native tags: - macOS: Finder xattrs (content-preserving, no re-upload) - Windows: XMP:Subject via ExifTool (re-upload expected once per file) Includes an interactive wizard that auto-detects the OS and Dropbox folder, walks the user through read-only token creation, previews tagged files, applies tags, and installs a background scheduler (launchd on macOS, Task Scheduler on Windows). Ships with: - 47-test suite (pytest) covering config, state DB, run_sync, writers - Gitea Actions CI for Linux / macOS / Windows with coverage reporting
This commit is contained in:
@@ -0,0 +1,122 @@
|
||||
"""Shared pytest fixtures and helpers."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# --- Isolate persistent state paths from the real user's home ---------------
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_user_state(tmp_path, monkeypatch):
|
||||
"""Redirect all per-user state files into a tmp dir.
|
||||
|
||||
Ensures tests never clobber real settings, tokens, or state DB on the
|
||||
developer's machine.
|
||||
"""
|
||||
fake_home = tmp_path / "home"
|
||||
fake_home.mkdir()
|
||||
monkeypatch.setattr(Path, "home", staticmethod(lambda: fake_home))
|
||||
|
||||
# Re-import config so its module-level paths re-evaluate against the
|
||||
# patched home. We clear and re-import to reset cached module constants.
|
||||
import importlib
|
||||
from tagsync import config
|
||||
|
||||
importlib.reload(config)
|
||||
yield
|
||||
# Leave things as-is after the test; tmp_path cleanup handles disk.
|
||||
|
||||
|
||||
# --- Fake Dropbox API objects ------------------------------------------------
|
||||
#
|
||||
# The Dropbox SDK may or may not be installed in the test env. We don't want
|
||||
# to require it, and the parts of core.py that need it are only reached via
|
||||
# helpers we can monkeypatch out.
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeFileMetadata:
|
||||
"""Stand-in for dropbox.files.FileMetadata."""
|
||||
name: str
|
||||
path_lower: str
|
||||
path_display: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeUserTag:
|
||||
tag_text: str
|
||||
|
||||
|
||||
class FakeTag:
|
||||
"""Stand-in for dropbox.files.Tag (union type)."""
|
||||
def __init__(self, tag_text: str) -> None:
|
||||
self._user = FakeUserTag(tag_text)
|
||||
|
||||
def get_user_generated_tag(self) -> FakeUserTag:
|
||||
return self._user
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakePathToTags:
|
||||
path: str
|
||||
tags: list
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeTagsResult:
|
||||
paths_to_tags: list
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patched_core(monkeypatch):
|
||||
"""Monkeypatch core's Dropbox-dependent helpers to controllable fakes.
|
||||
|
||||
Tests receive a `state` dict where they can configure:
|
||||
- `files`: list of FakeFileMetadata returned by _list_all_files
|
||||
- `tags_by_path`: dict[path_lower -> list[str]] returned by _fetch_tags
|
||||
"""
|
||||
from tagsync import core
|
||||
|
||||
state = {
|
||||
"files": [],
|
||||
"tags_by_path": {},
|
||||
"fetch_calls": 0,
|
||||
}
|
||||
|
||||
def fake_client(_token):
|
||||
return object() # placeholder; never used because list/fetch are faked
|
||||
|
||||
def fake_list(_dbx):
|
||||
return list(state["files"])
|
||||
|
||||
def fake_fetch(_dbx, files):
|
||||
state["fetch_calls"] += 1
|
||||
for f in files:
|
||||
tag_names = sorted(state["tags_by_path"].get(f.path_lower, []))
|
||||
yield f, tag_names
|
||||
|
||||
monkeypatch.setattr(core, "_dropbox_client", fake_client)
|
||||
monkeypatch.setattr(core, "_list_all_files", fake_list)
|
||||
monkeypatch.setattr(core, "_fetch_tags", fake_fetch)
|
||||
return state
|
||||
|
||||
|
||||
# --- Marker-based skips ------------------------------------------------------
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config, items):
|
||||
platform_skip = {
|
||||
"macos": sys.platform != "darwin",
|
||||
"windows": sys.platform != "win32",
|
||||
}
|
||||
for item in items:
|
||||
for mark_name, should_skip in platform_skip.items():
|
||||
if mark_name in item.keywords and should_skip:
|
||||
item.add_marker(
|
||||
pytest.mark.skip(reason=f"requires {mark_name}")
|
||||
)
|
||||
@@ -0,0 +1,114 @@
|
||||
"""Tests for tagsync.config: Settings persistence + token storage."""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_module():
|
||||
from tagsync import config
|
||||
return importlib.reload(config)
|
||||
|
||||
|
||||
# --- Settings ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_settings_defaults(config_module):
|
||||
s = config_module.Settings()
|
||||
assert s.local_root == ""
|
||||
assert s.platform == ""
|
||||
assert s.skip_online_only is True
|
||||
assert s.write_sidecars is False
|
||||
assert s.schedule_installed is False
|
||||
assert s.schedule_interval_minutes == 60
|
||||
|
||||
|
||||
def test_settings_roundtrip(config_module, tmp_path):
|
||||
path = tmp_path / "cfg.json"
|
||||
original = config_module.Settings(
|
||||
local_root="/home/u/Dropbox",
|
||||
platform="macos",
|
||||
skip_online_only=False,
|
||||
schedule_installed=True,
|
||||
schedule_interval_minutes=30,
|
||||
)
|
||||
original.save(path)
|
||||
|
||||
loaded = config_module.Settings.load(path)
|
||||
assert loaded == original
|
||||
|
||||
|
||||
def test_settings_load_missing_file_returns_defaults(config_module, tmp_path):
|
||||
loaded = config_module.Settings.load(tmp_path / "nope.json")
|
||||
assert loaded == config_module.Settings()
|
||||
|
||||
|
||||
def test_settings_load_ignores_unknown_keys(config_module, tmp_path):
|
||||
"""Older configs should load even if a new field was removed."""
|
||||
path = tmp_path / "cfg.json"
|
||||
path.write_text(json.dumps({
|
||||
"local_root": "/x",
|
||||
"platform": "macos",
|
||||
"future_field_we_dont_know_about": 42,
|
||||
}))
|
||||
loaded = config_module.Settings.load(path)
|
||||
assert loaded.local_root == "/x"
|
||||
assert loaded.platform == "macos"
|
||||
|
||||
|
||||
# --- Token storage (fallback file path) -------------------------------------
|
||||
|
||||
|
||||
def test_token_fallback_roundtrip(config_module, monkeypatch):
|
||||
"""Force the non-keyring fallback path and verify round-trip."""
|
||||
monkeypatch.setattr(config_module, "_keyring_available", lambda: False)
|
||||
|
||||
assert config_module.load_token() is None
|
||||
|
||||
location = config_module.save_token("sl.test-token-123")
|
||||
assert "plaintext" in location
|
||||
|
||||
assert config_module.load_token() == "sl.test-token-123"
|
||||
|
||||
config_module.clear_token()
|
||||
assert config_module.load_token() is None
|
||||
|
||||
|
||||
def test_token_fallback_file_permissions(config_module, monkeypatch):
|
||||
"""The fallback token file should be mode 0600 where supported."""
|
||||
import os
|
||||
import stat
|
||||
|
||||
monkeypatch.setattr(config_module, "_keyring_available", lambda: False)
|
||||
config_module.save_token("sl.perm-test")
|
||||
|
||||
mode = os.stat(config_module.TOKEN_FALLBACK_FILE).st_mode
|
||||
# Skip on Windows where POSIX modes aren't enforced the same way.
|
||||
if os.name == "posix":
|
||||
assert stat.S_IMODE(mode) == 0o600
|
||||
|
||||
|
||||
@pytest.mark.needs_keyring
|
||||
def test_token_keyring_roundtrip(config_module):
|
||||
"""If keyring is installed, verify we actually use it."""
|
||||
pytest.importorskip("keyring")
|
||||
|
||||
if not config_module._keyring_available():
|
||||
pytest.skip("keyring not importable")
|
||||
|
||||
import keyring
|
||||
# Use the in-memory fallback backend to avoid touching real Keychain.
|
||||
from keyring.backends import fail as _fail # noqa: F401 (ensure package)
|
||||
|
||||
try:
|
||||
import keyring.backends.null
|
||||
keyring.set_keyring(keyring.backends.null.Keyring())
|
||||
except Exception:
|
||||
pytest.skip("no safe keyring backend available for tests")
|
||||
|
||||
# Null keyring swallows set/get — just exercise code paths.
|
||||
config_module.save_token("sl.kr-test")
|
||||
config_module.clear_token()
|
||||
@@ -0,0 +1,268 @@
|
||||
"""Tests for tagsync.core: state DB + run_sync main loop.
|
||||
|
||||
Dropbox client calls are monkeypatched via the `patched_core` fixture, so
|
||||
these tests don't require network or the dropbox SDK to be installed.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tagsync import core
|
||||
from tests.conftest import FakeFileMetadata
|
||||
|
||||
|
||||
# --- State DB ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_open_state_creates_schema(tmp_path):
|
||||
db = tmp_path / "state.db"
|
||||
conn = core.open_state(db)
|
||||
try:
|
||||
tables = [
|
||||
row[0]
|
||||
for row in conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table'"
|
||||
)
|
||||
]
|
||||
finally:
|
||||
conn.close()
|
||||
assert "file_tags" in tables
|
||||
|
||||
|
||||
def test_open_state_is_idempotent(tmp_path):
|
||||
db = tmp_path / "state.db"
|
||||
core.open_state(db).close()
|
||||
# Re-opening an existing DB should not raise.
|
||||
core.open_state(db).close()
|
||||
|
||||
|
||||
def test_open_state_creates_parent_dir(tmp_path):
|
||||
db = tmp_path / "nested" / "dir" / "state.db"
|
||||
conn = core.open_state(db)
|
||||
conn.close()
|
||||
assert db.exists()
|
||||
|
||||
|
||||
# --- SyncStats --------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sync_stats_summary_format():
|
||||
s = core.SyncStats(applied=3, cleared=1, errors=0)
|
||||
summary = s.summary()
|
||||
for key in ("applied", "cleared", "skipped", "missing", "errors"):
|
||||
assert key in summary
|
||||
|
||||
|
||||
# --- Writer implementations used by run_sync --------------------------------
|
||||
|
||||
|
||||
class RecordingWriter:
|
||||
"""Fake Writer that records every call for assertions."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
supports_all: bool = True,
|
||||
online_paths: set[str] | None = None,
|
||||
fail_paths: set[str] | None = None,
|
||||
):
|
||||
self.supports_all = supports_all
|
||||
self.online_paths = online_paths or set()
|
||||
self.fail_paths = fail_paths or set()
|
||||
self.written: list[tuple[Path, list[str]]] = []
|
||||
|
||||
def supports(self, path: Path) -> bool:
|
||||
return self.supports_all
|
||||
|
||||
def is_online_only(self, path: Path) -> bool:
|
||||
return str(path) in self.online_paths
|
||||
|
||||
def write_tags(self, path: Path, tags: list[str]) -> None:
|
||||
if str(path) in self.fail_paths:
|
||||
raise RuntimeError("simulated tag failure")
|
||||
self.written.append((path, list(tags)))
|
||||
|
||||
|
||||
# --- run_sync happy paths ---------------------------------------------------
|
||||
|
||||
|
||||
def _make_local_file(local_root: Path, rel: str) -> Path:
|
||||
p = local_root / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.touch()
|
||||
return p
|
||||
|
||||
|
||||
def test_run_sync_applies_new_tags(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
_make_local_file(local_root, "Report.pdf")
|
||||
|
||||
patched_core["files"] = [
|
||||
FakeFileMetadata(name="Report.pdf", path_lower="/report.pdf",
|
||||
path_display="/Report.pdf"),
|
||||
]
|
||||
patched_core["tags_by_path"] = {"/report.pdf": ["urgent", "client"]}
|
||||
|
||||
writer = RecordingWriter()
|
||||
stats = core.run_sync(
|
||||
token="fake",
|
||||
local_root=local_root,
|
||||
state_db=tmp_path / "state.db",
|
||||
writer=writer,
|
||||
)
|
||||
assert stats.applied == 1
|
||||
assert stats.errors == 0
|
||||
assert writer.written == [
|
||||
(local_root / "Report.pdf", ["client", "urgent"]), # sorted
|
||||
]
|
||||
|
||||
|
||||
def test_run_sync_skips_unchanged_tags(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
_make_local_file(local_root, "a.pdf")
|
||||
|
||||
patched_core["files"] = [
|
||||
FakeFileMetadata("a.pdf", "/a.pdf", "/a.pdf")
|
||||
]
|
||||
patched_core["tags_by_path"] = {"/a.pdf": ["x"]}
|
||||
|
||||
writer = RecordingWriter()
|
||||
state_db = tmp_path / "state.db"
|
||||
|
||||
first = core.run_sync("t", local_root, state_db, writer)
|
||||
assert first.applied == 1
|
||||
|
||||
# Second run with the same Dropbox state should skip.
|
||||
second = core.run_sync("t", local_root, state_db, writer)
|
||||
assert second.applied == 0
|
||||
assert second.skipped == 1
|
||||
assert len(writer.written) == 1 # unchanged from first run
|
||||
|
||||
|
||||
def test_run_sync_clears_tags_when_removed_upstream(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
_make_local_file(local_root, "b.pdf")
|
||||
|
||||
patched_core["files"] = [FakeFileMetadata("b.pdf", "/b.pdf", "/b.pdf")]
|
||||
patched_core["tags_by_path"] = {"/b.pdf": ["temp"]}
|
||||
|
||||
writer = RecordingWriter()
|
||||
state_db = tmp_path / "state.db"
|
||||
core.run_sync("t", local_root, state_db, writer)
|
||||
|
||||
# Upstream removes the tag.
|
||||
patched_core["tags_by_path"] = {"/b.pdf": []}
|
||||
stats = core.run_sync("t", local_root, state_db, writer)
|
||||
assert stats.cleared == 1
|
||||
assert writer.written[-1] == (local_root / "b.pdf", [])
|
||||
|
||||
|
||||
def test_run_sync_skips_missing_local_files(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
# Don't create the local file.
|
||||
|
||||
patched_core["files"] = [FakeFileMetadata("c.pdf", "/c.pdf", "/c.pdf")]
|
||||
patched_core["tags_by_path"] = {"/c.pdf": ["x"]}
|
||||
|
||||
writer = RecordingWriter()
|
||||
stats = core.run_sync(
|
||||
"t", local_root, tmp_path / "state.db", writer
|
||||
)
|
||||
assert stats.missing == 1
|
||||
assert writer.written == []
|
||||
|
||||
|
||||
def test_run_sync_respects_unsupported_files(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
_make_local_file(local_root, "data.zip")
|
||||
|
||||
patched_core["files"] = [
|
||||
FakeFileMetadata("data.zip", "/data.zip", "/data.zip")
|
||||
]
|
||||
patched_core["tags_by_path"] = {"/data.zip": ["archive"]}
|
||||
|
||||
writer = RecordingWriter(supports_all=False)
|
||||
stats = core.run_sync(
|
||||
"t", local_root, tmp_path / "state.db", writer
|
||||
)
|
||||
assert stats.unsupported == 1
|
||||
assert stats.applied == 0
|
||||
|
||||
|
||||
def test_run_sync_skips_online_only_files_when_requested(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
local_file = _make_local_file(local_root, "cloud.pdf")
|
||||
|
||||
patched_core["files"] = [FakeFileMetadata("cloud.pdf", "/cloud.pdf", "/cloud.pdf")]
|
||||
patched_core["tags_by_path"] = {"/cloud.pdf": ["tag"]}
|
||||
|
||||
writer = RecordingWriter(online_paths={str(local_file)})
|
||||
stats = core.run_sync(
|
||||
"t", local_root, tmp_path / "state.db", writer,
|
||||
skip_online_only=True,
|
||||
)
|
||||
assert stats.online_only == 1
|
||||
assert stats.applied == 0
|
||||
|
||||
|
||||
def test_run_sync_dry_run_touches_nothing(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
_make_local_file(local_root, "d.pdf")
|
||||
|
||||
patched_core["files"] = [FakeFileMetadata("d.pdf", "/d.pdf", "/d.pdf")]
|
||||
patched_core["tags_by_path"] = {"/d.pdf": ["x"]}
|
||||
|
||||
writer = RecordingWriter()
|
||||
stats = core.run_sync(
|
||||
"t", local_root, tmp_path / "state.db", writer, dry_run=True,
|
||||
)
|
||||
assert stats.applied == 1
|
||||
assert writer.written == [] # nothing actually written
|
||||
|
||||
|
||||
def test_run_sync_counts_writer_errors(tmp_path, patched_core):
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
local_file = _make_local_file(local_root, "e.pdf")
|
||||
|
||||
patched_core["files"] = [FakeFileMetadata("e.pdf", "/e.pdf", "/e.pdf")]
|
||||
patched_core["tags_by_path"] = {"/e.pdf": ["x"]}
|
||||
|
||||
writer = RecordingWriter(fail_paths={str(local_file)})
|
||||
stats = core.run_sync(
|
||||
"t", local_root, tmp_path / "state.db", writer,
|
||||
)
|
||||
assert stats.errors == 1
|
||||
assert stats.applied == 0
|
||||
|
||||
|
||||
# --- Path translation -------------------------------------------------------
|
||||
|
||||
|
||||
def test_path_display_to_local(tmp_path, patched_core):
|
||||
"""Dropbox's /A/B/C.pdf must map under the configured local root."""
|
||||
local_root = tmp_path / "Dropbox"
|
||||
local_root.mkdir()
|
||||
_make_local_file(local_root, "A/B/C.pdf")
|
||||
|
||||
patched_core["files"] = [
|
||||
FakeFileMetadata("C.pdf", "/a/b/c.pdf", "/A/B/C.pdf")
|
||||
]
|
||||
patched_core["tags_by_path"] = {"/a/b/c.pdf": ["t"]}
|
||||
|
||||
writer = RecordingWriter()
|
||||
core.run_sync(
|
||||
"t", local_root, tmp_path / "state.db", writer,
|
||||
)
|
||||
assert writer.written
|
||||
written_path, _ = writer.written[0]
|
||||
assert written_path == local_root / "A" / "B" / "C.pdf"
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Tests for tagsync.detect: platform detection + Dropbox folder probing."""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def detect_module():
|
||||
from tagsync import detect
|
||||
return importlib.reload(detect)
|
||||
|
||||
|
||||
def test_detect_platform_returns_known_value(detect_module):
|
||||
assert detect_module.detect_platform() in {"macos", "windows", "linux"}
|
||||
|
||||
|
||||
def test_platform_label_nonempty(detect_module):
|
||||
assert detect_module.platform_label()
|
||||
|
||||
|
||||
def test_python_label_nonempty(detect_module):
|
||||
label = detect_module.python_label()
|
||||
assert label.startswith("Python ")
|
||||
|
||||
|
||||
def test_has_module_true_for_stdlib(detect_module):
|
||||
assert detect_module.has_module("json")
|
||||
assert detect_module.has_module("pathlib")
|
||||
|
||||
|
||||
def test_has_module_false_for_garbage(detect_module):
|
||||
assert not detect_module.has_module("definitely_not_a_real_module_xyz")
|
||||
|
||||
|
||||
def test_detect_dropbox_folder_from_info_json(detect_module, tmp_path, monkeypatch):
|
||||
"""When Dropbox's info.json exists, we should parse its path field."""
|
||||
dropbox_dir = tmp_path / "Dropbox"
|
||||
dropbox_dir.mkdir()
|
||||
info_dir = tmp_path / ".dropbox"
|
||||
info_dir.mkdir()
|
||||
info = info_dir / "info.json"
|
||||
info.write_text(json.dumps({
|
||||
"personal": {"path": str(dropbox_dir), "host": 12345},
|
||||
}))
|
||||
|
||||
monkeypatch.setattr(
|
||||
detect_module,
|
||||
"_dropbox_info_candidates",
|
||||
lambda: [info],
|
||||
)
|
||||
result = detect_module.detect_dropbox_folder()
|
||||
assert result == dropbox_dir
|
||||
|
||||
|
||||
def test_detect_dropbox_folder_handles_malformed_json(detect_module, tmp_path, monkeypatch):
|
||||
info = tmp_path / "info.json"
|
||||
info.write_text("not valid json {{{")
|
||||
|
||||
monkeypatch.setattr(
|
||||
detect_module,
|
||||
"_dropbox_info_candidates",
|
||||
lambda: [info],
|
||||
)
|
||||
# Should fall back silently, not raise.
|
||||
result = detect_module.detect_dropbox_folder()
|
||||
# No fallback path in tmp, so None is acceptable.
|
||||
assert result is None or result.is_dir()
|
||||
|
||||
|
||||
def test_detect_dropbox_folder_returns_none_if_missing(detect_module, monkeypatch):
|
||||
from pathlib import Path
|
||||
monkeypatch.setattr(detect_module, "_dropbox_info_candidates", lambda: [])
|
||||
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/definitely/does/not/exist")))
|
||||
assert detect_module.detect_dropbox_folder() is None
|
||||
|
||||
|
||||
def test_detect_dropbox_folder_prefers_personal_over_business(detect_module, tmp_path, monkeypatch):
|
||||
personal = tmp_path / "Dropbox-Personal"
|
||||
business = tmp_path / "Dropbox-Business"
|
||||
personal.mkdir()
|
||||
business.mkdir()
|
||||
info = tmp_path / "info.json"
|
||||
info.write_text(json.dumps({
|
||||
"personal": {"path": str(personal)},
|
||||
"business": {"path": str(business)},
|
||||
}))
|
||||
monkeypatch.setattr(
|
||||
detect_module,
|
||||
"_dropbox_info_candidates",
|
||||
lambda: [info],
|
||||
)
|
||||
assert detect_module.detect_dropbox_folder() == personal
|
||||
|
||||
|
||||
def test_find_exiftool_uses_shutil_which(detect_module, monkeypatch):
|
||||
import shutil
|
||||
monkeypatch.setattr(shutil, "which", lambda name: "/usr/bin/exiftool" if name == "exiftool" else None)
|
||||
result = detect_module.find_exiftool()
|
||||
assert result is not None
|
||||
assert result.name == "exiftool"
|
||||
|
||||
|
||||
def test_find_exiftool_returns_none_when_absent(detect_module, monkeypatch):
|
||||
import shutil
|
||||
monkeypatch.setattr(shutil, "which", lambda _name: None)
|
||||
assert detect_module.find_exiftool() is None
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Tests for tagsync.platform_macos.
|
||||
|
||||
The plist encoding test runs anywhere. The xattr round-trip runs only on
|
||||
systems where the extended-attribute syscalls are available (macOS and
|
||||
most Linux filesystems).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import plistlib
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tagsync.platform_macos import FINDER_TAGS_XATTR, MacWriter
|
||||
|
||||
|
||||
def test_mac_writer_supports_all_file_types():
|
||||
w = MacWriter()
|
||||
for path in ("a.jpg", "b.zip", "c.txt", "noextension"):
|
||||
assert w.supports(Path(path))
|
||||
|
||||
|
||||
def test_mac_writer_is_online_only_is_false():
|
||||
w = MacWriter()
|
||||
assert w.is_online_only(Path("/any/file")) is False
|
||||
|
||||
|
||||
def test_plist_encoding_matches_finder_format():
|
||||
"""Finder expects a binary plist of a string array."""
|
||||
from tagsync.platform_macos import plistlib # re-export
|
||||
tags = ["alpha", "beta", "gamma"]
|
||||
encoded = plistlib.dumps(tags, fmt=plistlib.FMT_BINARY)
|
||||
# Binary plists start with 'bplist'.
|
||||
assert encoded.startswith(b"bplist")
|
||||
decoded = plistlib.loads(encoded)
|
||||
assert decoded == tags
|
||||
|
||||
|
||||
def _xattr_supported(path: Path) -> bool:
|
||||
"""Can we setxattr / getxattr on files in this directory?"""
|
||||
if not hasattr(os, "setxattr") or not hasattr(os, "getxattr"):
|
||||
return False
|
||||
try:
|
||||
# Use a neutral namespace; on Linux, user.* always works on ext4.
|
||||
os.setxattr(path, "user.tagsync_probe", b"ok")
|
||||
val = os.getxattr(path, "user.tagsync_probe")
|
||||
os.removexattr(path, "user.tagsync_probe")
|
||||
return val == b"ok"
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def xattr_capable_file(tmp_path):
|
||||
f = tmp_path / "target"
|
||||
f.touch()
|
||||
if not _xattr_supported(f):
|
||||
pytest.skip("filesystem does not support extended attributes")
|
||||
return f
|
||||
|
||||
|
||||
def test_write_and_read_tags_via_xattr(xattr_capable_file):
|
||||
"""Round-trip the actual xattr the script uses in production.
|
||||
|
||||
On Linux, the 'com.apple.metadata:*' namespace may not be allowed,
|
||||
so this test uses the MacWriter only on Darwin.
|
||||
"""
|
||||
if sys.platform != "darwin":
|
||||
pytest.skip("Finder xattr namespace only writable on macOS")
|
||||
|
||||
w = MacWriter()
|
||||
w.write_tags(xattr_capable_file, ["alpha", "beta"])
|
||||
|
||||
raw = os.getxattr(xattr_capable_file, FINDER_TAGS_XATTR)
|
||||
assert plistlib.loads(raw) == ["alpha", "beta"]
|
||||
|
||||
|
||||
def test_clear_tags_via_xattr(xattr_capable_file):
|
||||
if sys.platform != "darwin":
|
||||
pytest.skip("Finder xattr namespace only writable on macOS")
|
||||
|
||||
w = MacWriter()
|
||||
w.write_tags(xattr_capable_file, ["alpha"])
|
||||
w.write_tags(xattr_capable_file, [])
|
||||
|
||||
with pytest.raises(OSError):
|
||||
os.getxattr(xattr_capable_file, FINDER_TAGS_XATTR)
|
||||
|
||||
|
||||
def test_clear_tags_when_none_set_is_noop(tmp_path):
|
||||
"""Removing a non-existent xattr should not raise."""
|
||||
if sys.platform != "darwin":
|
||||
pytest.skip("requires macOS xattr semantics")
|
||||
f = tmp_path / "fresh"
|
||||
f.touch()
|
||||
MacWriter().write_tags(f, []) # should not raise
|
||||
|
||||
|
||||
# --- launchd plist generation -----------------------------------------------
|
||||
|
||||
|
||||
def test_install_schedule_writes_plist(tmp_path, monkeypatch):
|
||||
"""We don't want to actually call launchctl; intercept it and verify
|
||||
the plist contents."""
|
||||
import tagsync.platform_macos as pm
|
||||
|
||||
calls = []
|
||||
|
||||
class FakeResult:
|
||||
returncode = 0
|
||||
stderr = ""
|
||||
stdout = ""
|
||||
|
||||
monkeypatch.setattr(pm.subprocess, "run", lambda args, **kw: calls.append(args) or FakeResult())
|
||||
|
||||
# Redirect plist + log paths into tmp to avoid touching real HOME.
|
||||
fake_plist = tmp_path / "agent.plist"
|
||||
fake_log = tmp_path / "log"
|
||||
monkeypatch.setattr(pm, "LAUNCHD_PLIST", fake_plist)
|
||||
monkeypatch.setattr(pm, "LOG_PATH", fake_log)
|
||||
|
||||
entry = tmp_path / "tagsync.py"
|
||||
entry.write_text("# placeholder")
|
||||
|
||||
result_path = pm.install_schedule(entry, interval_minutes=30)
|
||||
assert result_path == fake_plist
|
||||
content = fake_plist.read_text()
|
||||
assert "StartInterval" in content
|
||||
assert "1800" in content # 30 min -> 1800 sec
|
||||
assert str(entry) in content
|
||||
assert "launchctl" in " ".join(" ".join(c) for c in calls)
|
||||
|
||||
|
||||
def test_uninstall_schedule_when_not_installed(tmp_path, monkeypatch):
|
||||
import tagsync.platform_macos as pm
|
||||
|
||||
monkeypatch.setattr(pm, "LAUNCHD_PLIST", tmp_path / "missing.plist")
|
||||
assert pm.uninstall_schedule() is False
|
||||
@@ -0,0 +1,154 @@
|
||||
"""Tests for tagsync.platform_windows: sidecar writer + support matrix.
|
||||
|
||||
Runs on any OS — the Windows-specific online-only stat test is skipped
|
||||
except on Windows.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tagsync.platform_windows import (
|
||||
XMP_WRITABLE_EXTS,
|
||||
WindowsWriter,
|
||||
)
|
||||
|
||||
|
||||
# --- supports() --------------------------------------------------------------
|
||||
|
||||
|
||||
def test_supports_xmp_formats():
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=False)
|
||||
for ext in (".jpg", ".pdf", ".docx", ".mp4"):
|
||||
assert w.supports(Path(f"foo{ext}")), f"expected support for {ext}"
|
||||
|
||||
|
||||
def test_supports_rejects_unknown_without_sidecars():
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=False)
|
||||
assert not w.supports(Path("archive.zip"))
|
||||
assert not w.supports(Path("notes.txt"))
|
||||
|
||||
|
||||
def test_supports_accepts_unknown_with_sidecars():
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=True)
|
||||
assert w.supports(Path("archive.zip"))
|
||||
assert w.supports(Path("notes.txt"))
|
||||
|
||||
|
||||
def test_supports_is_case_insensitive():
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=False)
|
||||
assert w.supports(Path("IMAGE.JPG"))
|
||||
assert w.supports(Path("report.PDF"))
|
||||
|
||||
|
||||
# --- sidecar writer ---------------------------------------------------------
|
||||
|
||||
|
||||
def test_sidecar_written_for_unsupported_type(tmp_path):
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=True)
|
||||
target = tmp_path / "archive.zip"
|
||||
target.touch()
|
||||
|
||||
w.write_tags(target, ["project-alpha", "urgent"])
|
||||
|
||||
sidecar = tmp_path / "archive.zip.tags.json"
|
||||
assert sidecar.exists()
|
||||
data = json.loads(sidecar.read_text())
|
||||
assert data == {"tags": ["project-alpha", "urgent"]}
|
||||
|
||||
|
||||
def test_sidecar_cleared_when_tags_empty(tmp_path):
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=True)
|
||||
target = tmp_path / "archive.zip"
|
||||
target.touch()
|
||||
(tmp_path / "archive.zip.tags.json").write_text('{"tags": ["x"]}')
|
||||
|
||||
w.write_tags(target, [])
|
||||
assert not (tmp_path / "archive.zip.tags.json").exists()
|
||||
|
||||
|
||||
def test_sidecar_clear_nonexistent_is_ok(tmp_path):
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=True)
|
||||
target = tmp_path / "archive.zip"
|
||||
target.touch()
|
||||
# Should not raise even though no sidecar exists.
|
||||
w.write_tags(target, [])
|
||||
|
||||
|
||||
def test_unsupported_without_sidecars_raises(tmp_path):
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=False)
|
||||
target = tmp_path / "archive.zip"
|
||||
target.touch()
|
||||
with pytest.raises(RuntimeError):
|
||||
w.write_tags(target, ["tag"])
|
||||
|
||||
|
||||
# --- exiftool invocation ----------------------------------------------------
|
||||
#
|
||||
# We don't actually run exiftool; monkeypatch subprocess.run to capture args.
|
||||
|
||||
|
||||
def test_exiftool_args_are_well_formed(tmp_path, monkeypatch):
|
||||
captured = {}
|
||||
|
||||
def fake_run(args, **kwargs):
|
||||
captured["args"] = args
|
||||
class R:
|
||||
returncode = 0
|
||||
stderr = ""
|
||||
return R()
|
||||
|
||||
import tagsync.platform_windows as pw
|
||||
monkeypatch.setattr(pw.subprocess, "run", fake_run)
|
||||
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=False)
|
||||
target = tmp_path / "pic.jpg"
|
||||
target.touch()
|
||||
w.write_tags(target, ["alpha", "beta"])
|
||||
|
||||
args = captured["args"]
|
||||
assert args[0] == "exiftool"
|
||||
assert "-overwrite_original" in args
|
||||
# Subject should be cleared then appended for each tag.
|
||||
assert "-XMP-dc:Subject=" in args
|
||||
assert "-XMP-dc:Subject+=alpha" in args
|
||||
assert "-XMP-dc:Subject+=beta" in args
|
||||
assert str(target) in args
|
||||
|
||||
|
||||
def test_exiftool_failure_raises(tmp_path, monkeypatch):
|
||||
def fake_run(args, **kwargs):
|
||||
class R:
|
||||
returncode = 1
|
||||
stderr = "bad things happened"
|
||||
return R()
|
||||
|
||||
import tagsync.platform_windows as pw
|
||||
monkeypatch.setattr(pw.subprocess, "run", fake_run)
|
||||
|
||||
w = WindowsWriter(exiftool="exiftool", write_sidecars=False)
|
||||
target = tmp_path / "pic.jpg"
|
||||
target.touch()
|
||||
with pytest.raises(RuntimeError, match="bad things happened"):
|
||||
w.write_tags(target, ["x"])
|
||||
|
||||
|
||||
# --- online-only detection --------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.windows
|
||||
def test_is_online_only_returns_false_for_normal_files(tmp_path):
|
||||
w = WindowsWriter(exiftool="exiftool")
|
||||
f = tmp_path / "regular.txt"
|
||||
f.touch()
|
||||
assert w.is_online_only(f) is False
|
||||
|
||||
|
||||
def test_is_online_only_returns_false_when_attribute_unavailable(tmp_path):
|
||||
"""On non-Windows, st_file_attributes doesn't exist -> should return False."""
|
||||
w = WindowsWriter(exiftool="exiftool")
|
||||
f = tmp_path / "regular.txt"
|
||||
f.touch()
|
||||
assert w.is_online_only(f) is False
|
||||
Reference in New Issue
Block a user