From e3630200804a6291982c77bba12e3cf0b45a3ca2 Mon Sep 17 00:00:00 2001 From: Jared Knowles Date: Mon, 20 Apr 2026 14:16:50 -0400 Subject: [PATCH] feat: initial Dropbox tag sync tool 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 --- .gitea/workflows/ci.yml | 95 +++++++ .gitignore | 13 + README.md | 197 ++++++++++++++ pyproject.toml | 27 ++ requirements-dev.txt | 3 + requirements.txt | 2 + tagsync.py | 214 ++++++++++++++++ tagsync/__init__.py | 3 + tagsync/config.py | 94 +++++++ tagsync/core.py | 267 +++++++++++++++++++ tagsync/detect.py | 95 +++++++ tagsync/platform_macos.py | 125 +++++++++ tagsync/platform_windows.py | 129 ++++++++++ tagsync/wizard.py | 455 +++++++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/conftest.py | 122 +++++++++ tests/test_config.py | 114 +++++++++ tests/test_core.py | 268 +++++++++++++++++++ tests/test_detect.py | 109 ++++++++ tests/test_platform_macos.py | 140 ++++++++++ tests/test_platform_windows.py | 154 +++++++++++ 21 files changed, 2626 insertions(+) create mode 100644 .gitea/workflows/ci.yml create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt create mode 100644 requirements.txt create mode 100644 tagsync.py create mode 100644 tagsync/__init__.py create mode 100644 tagsync/config.py create mode 100644 tagsync/core.py create mode 100644 tagsync/detect.py create mode 100644 tagsync/platform_macos.py create mode 100644 tagsync/platform_windows.py create mode 100644 tagsync/wizard.py create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_config.py create mode 100644 tests/test_core.py create mode 100644 tests/test_detect.py create mode 100644 tests/test_platform_macos.py create mode 100644 tests/test_platform_windows.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..93a5e4d --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,95 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test-linux: + name: Tests (Linux, Python ${{ matrix.python-version }}) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.9", "3.11", "3.12"] + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: | + requirements.txt + requirements-dev.txt + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements-dev.txt + + - name: Syntax check + run: | + python -m py_compile tagsync.py + python -c "import pathlib, py_compile; [py_compile.compile(str(p), doraise=True) for p in pathlib.Path('tagsync').rglob('*.py')]" + + - name: CLI smoke test + run: | + python tagsync.py --help + python tagsync.py status + + - name: Run tests with coverage + run: | + pytest --cov=tagsync --cov-report=term-missing --cov-report=xml + + - name: Upload coverage artifact + if: matrix.python-version == '3.12' + uses: actions/upload-artifact@v3 + with: + name: coverage-xml + path: coverage.xml + retention-days: 14 + + test-macos: + name: Tests (macOS) + runs-on: macos-latest + # macOS runners are more expensive — only run on main-branch pushes + # and PRs that touch platform_macos.py or core logic. + if: github.event_name == 'push' || contains(github.event.pull_request.labels.*.name, 'platform') + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install dependencies + run: pip install -r requirements-dev.txt + + - name: Run tests + run: pytest -v + + test-windows: + name: Tests (Windows) + runs-on: windows-latest + if: github.event_name == 'push' || contains(github.event.pull_request.labels.*.name, 'platform') + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install dependencies + run: pip install -r requirements-dev.txt + + - name: Run tests + run: pytest -v diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7630ba9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ +.coverage +coverage.xml +htmlcov/ +*.egg-info/ +build/ +dist/ +.venv/ +venv/ +.env diff --git a/README.md b/README.md new file mode 100644 index 0000000..b197db1 --- /dev/null +++ b/README.md @@ -0,0 +1,197 @@ +# Dropbox Tag Sync + +Bring Dropbox web tags down to your computer so you can search for tagged +files in Finder (macOS) or File Explorer (Windows) as if they were +first-class local tags. + +## The problem this solves + +Dropbox lets you add **tags** to files from its web interface. Those tags +live in Dropbox's cloud metadata — they are *not* stored inside the files +themselves. The Dropbox desktop app syncs file **contents**, but it does +not copy tags onto your local files. As a result, tags you rely on for +search and organization show up on dropbox.com but are invisible to +Finder, Spotlight, File Explorer, and Windows Search. + +This tool closes that gap. It connects to Dropbox, reads each file's +tags, and writes them to your local files in a form your operating +system treats as a first-class tag. + +## Quick start + +``` +pip install -r requirements.txt +python tagsync.py wizard +``` + +That's it. The wizard detects your OS, finds your Dropbox folder, walks +you through creating a read-only Dropbox access token, previews what +will be tagged, applies the tags, and offers to install a background +scheduler so new tags keep flowing down automatically. + +Everything after that is optional — if you want to re-run a sync, +check status, or manage the schedule, use the other subcommands below. + +## Requirements + +- **Python 3.9+** +- The packages in `requirements.txt` (`pip install -r requirements.txt`) +- **Windows only:** [ExifTool](https://exiftool.org/) on your `PATH` + +## What's in this repo + +| File / folder | Purpose | +| ------------------------ | ------------------------------------------ | +| `tagsync.py` | Single CLI entry point | +| `tagsync/` | Package with all the logic | +| `requirements.txt` | Python dependencies | +| `README.md` | This document | + +Inside `tagsync/`: + +| File | Purpose | +| ------------------------- | --------------------------------------------- | +| `wizard.py` | Interactive setup flow | +| `core.py` | Dropbox listing, tag fetching, sync loop | +| `config.py` | Settings file + secure token storage | +| `detect.py` | OS / Dropbox folder / dependency detection | +| `platform_macos.py` | Finder-xattr writer + launchd scheduler | +| `platform_windows.py` | ExifTool-based XMP writer + Task Scheduler | + +## Commands + +All commands are run as `python tagsync.py `. + +| Command | What it does | +| ------------------------------ | ---------------------------------------------------- | +| `wizard` | Interactive setup (run this first) | +| `run` | Perform a one-shot sync using saved settings | +| `run --dry-run` | Show what would change without touching any files | +| `status` | Show config, token presence, and schedule state | +| `schedule install` | Install the background scheduled task | +| `schedule install --interval 30` | Install, running every 30 minutes | +| `schedule uninstall` | Remove the background scheduled task | +| `schedule status` | Show whether the scheduler is installed and loaded | +| `reset --all` | Clear stored token, settings, and local state DB | +| `reset --token` | Clear only the token | + +## How it works + +1. You create a Dropbox access token with **read-only** metadata scope. + The wizard walks you through this. The token lives in your OS + keyring (macOS Keychain or Windows Credential Manager). +2. The tool lists every file in your Dropbox and asks Dropbox which + tags each file has. +3. For files whose tags have changed since the last run, it writes + those tags to the local copy. +4. A small SQLite cache remembers what was written so subsequent runs + only touch files whose tags actually changed. + +Nothing is ever deleted from Dropbox. The tool only **reads** from +Dropbox and **writes** to local files on your machine. + +## The important platform difference + +macOS and Windows store "file tags" in fundamentally different ways, +and that shapes what happens under the hood. + +### macOS — clean and invisible + +macOS has a universal, built-in tag system. Tags are stored in an +**extended attribute** (`xattr`) alongside the file, not inside it. The +tool writes the xattr directly. Consequences: + +- Every file type can be tagged (text, zip, source code, anything). +- The file's contents are untouched, so Dropbox does **not** see the + file as modified and does **not** re-upload it. +- The file's modification time does not change. + +### Windows — embeds tags into the file + +Windows has no universal tag store. The "Tags" field in Explorer is +actually the **XMP:Subject** metadata field embedded inside the file. +This is a limitation of the operating system, not the tool. + +Consequences: + +- Only file formats that support XMP (images, PDFs, Office docs, most + audio and video) can be tagged. The tool skips unsupported formats + by default and can optionally write a `.tags.json` sidecar next to + them instead. +- Because the tag lives inside the file, writing it **does** modify + the file. Dropbox will notice and re-upload the file once. After + that first sync, tags are stored in both the cloud and the local + copy, and future runs only touch files whose Dropbox tags change + again. +- If Dropbox Smart Sync leaves some files online-only, tagging one + would trigger a download. The tool can skip online-only files (on + by default in the wizard). + +> **Plan for the first Windows run.** If there are many tagged files, +> the first run will queue a large number of re-uploads as each file +> gets its tags embedded. Schedule it overnight, on a good connection, +> and expect a one-time spike in Dropbox activity. Subsequent runs are +> quiet. + +## One-time Dropbox app setup + +The wizard will open this page and walk you through it, but for +reference: + +1. Go to and click **Create app**. +2. Choose: + - **Scoped access** + - **Full Dropbox** + - A name of your choosing (e.g. `tag-sync`) +3. Open the **Permissions** tab and enable: + - `files.metadata.read` +4. Click **Submit**. +5. On the **Settings** tab, scroll to **OAuth 2 → Generated access token** + and click **Generate**. Paste that token into the wizard. + +The generated token only has permission to read file metadata. It +cannot modify anything in Dropbox even if it leaked. + +## Where things are stored + +| Item | Location | +| ------------------ | ---------------------------------------------------- | +| Token (preferred) | macOS Keychain / Windows Credential Manager | +| Token (fallback) | `~/.dropbox_tag_sync/token` (mode 0600) | +| Settings | `~/.dropbox_tag_sync/config.json` | +| Local state cache | `~/.dropbox_tag_sync/state.db` | +| Log (macOS) | `~/Library/Logs/dropbox-tag-sync.log` | +| launchd plist | `~/Library/LaunchAgents/com.dropboxtagsync.agent.plist` | +| Windows task | Task Scheduler, name: `DropboxTagSync` | + +## Troubleshooting + +**"not found locally"** — the file exists in Dropbox but hasn't synced +down yet. Let Dropbox finish syncing, or (on Windows) mark the file +as "always keep on this device" if it's online-only. + +**Windows tags don't show up in Explorer after a run** — Explorer's +Tags column may be hidden; right-click a column header and add +**Tags**. Windows Search may need a few minutes to re-index. + +**"exiftool not found"** — install it from +and confirm `exiftool -ver` works in a terminal. + +**The state cache got into a weird state** — run +`python tagsync.py reset --state`. The next run will re-examine every +file; no data is lost. + +**The token is wrong or revoked** — run +`python tagsync.py reset --token`, then `python tagsync.py wizard` +to enter a new one. + +## Safety summary + +- **Read-only** with Dropbox (token scope: `files.metadata.read`). +- **Non-destructive on macOS**: only extended attributes are written. +- **Content-modifying on Windows**: rewrites the XMP metadata section + of supported files (one-time re-upload per tagged file is expected). +- **Idempotent**: safe to run as often as you like. +- **Resumable**: if interrupted, the next run picks up where it left off. +- **One-way**: tags flow cloud → local only. Tagging on the web is + still the source of truth. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f8b18fb --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,27 @@ +[tool.pytest.ini_options] +minversion = "7.0" +testpaths = ["tests"] +addopts = [ + "-ra", + "--strict-markers", + "--strict-config", +] +markers = [ + "macos: tests that only run on macOS", + "windows: tests that only run on Windows", + "needs_dropbox: tests that require the dropbox SDK installed", + "needs_keyring: tests that require the keyring package installed", +] + +[tool.coverage.run] +source = ["tagsync"] +branch = true +omit = ["tests/*"] + +[tool.coverage.report] +exclude_lines = [ + "pragma: no cover", + "raise NotImplementedError", + "if TYPE_CHECKING:", + "if __name__ == .__main__.:", +] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..db656b4 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pytest>=7.0 +pytest-cov>=4.0 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6385dce --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +dropbox>=11.36.0 +keyring>=24.0.0 diff --git a/tagsync.py b/tagsync.py new file mode 100644 index 0000000..b79aa1e --- /dev/null +++ b/tagsync.py @@ -0,0 +1,214 @@ +#!/usr/bin/env python3 +"""Dropbox Tag Sync — CLI entry point. + +Subcommands: + wizard Interactive setup (run this first) + run Perform a one-shot sync using saved settings + status Show current configuration and schedule state + schedule Install/uninstall the background scheduled task + reset Clear stored token and/or settings +""" +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from tagsync import config, core, detect, wizard + +ENTRY_SCRIPT = Path(__file__).resolve() + + +def cmd_wizard(_args: argparse.Namespace) -> int: + return wizard.run_wizard(entry_script=ENTRY_SCRIPT) + + +def cmd_run(args: argparse.Namespace) -> int: + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + ) + + settings = config.Settings.load() + token = config.load_token() + + if not settings.local_root or not token: + print( + "Not configured yet. Run the wizard first:\n" + f" {sys.executable} {ENTRY_SCRIPT} wizard", + file=sys.stderr, + ) + return 2 + + local_root = Path(settings.local_root) + if not local_root.is_dir(): + print( + f"Configured Dropbox folder no longer exists: {local_root}\n" + "Re-run the wizard to update it.", + file=sys.stderr, + ) + return 2 + + writer = wizard._build_writer(settings) + stats = core.run_sync( + token=token, + local_root=local_root, + state_db=config.STATE_DB_FILE, + writer=writer, + skip_online_only=settings.skip_online_only, + dry_run=args.dry_run, + ) + print(f"done: {stats.summary()}") + return 0 if stats.errors == 0 else 1 + + +def cmd_status(_args: argparse.Namespace) -> int: + settings = config.Settings.load() + token_present = config.load_token() is not None + + print("Dropbox Tag Sync — status") + print("-" * 40) + print(f"Configured: {'yes' if settings.local_root else 'no (run wizard)'}") + print(f"Platform: {settings.platform or detect.detect_platform()}") + print(f"Dropbox folder: {settings.local_root or '(not set)'}") + print(f"Token stored: {'yes' if token_present else 'no'}") + if settings.platform == "windows": + print(f"ExifTool: {settings.exiftool or '(not set)'}") + print(f"Skip online-only:{settings.skip_online_only}") + print(f"Sidecars: {settings.write_sidecars}") + + sched = _schedule_status(settings.platform) + print() + print("Schedule") + for key, val in sched.items(): + print(f" {key}: {val}") + + print() + print(f"State DB: {config.STATE_DB_FILE}") + return 0 + + +def cmd_schedule(args: argparse.Namespace) -> int: + settings = config.Settings.load() + plat = settings.platform or detect.detect_platform() + + if args.action == "install": + interval = args.interval or settings.schedule_interval_minutes or 60 + try: + if plat == "macos": + from tagsync import platform_macos + plist = platform_macos.install_schedule(ENTRY_SCRIPT, interval_minutes=interval) + print(f"Installed launchd agent: {plist}") + elif plat == "windows": + from tagsync import platform_windows + platform_windows.install_schedule(ENTRY_SCRIPT, interval_minutes=interval) + print(f"Installed scheduled task: {platform_windows.TASK_NAME}") + else: + print(f"Unsupported platform: {plat}", file=sys.stderr) + return 2 + except Exception as e: + print(f"Schedule install failed: {e}", file=sys.stderr) + return 1 + settings.schedule_installed = True + settings.schedule_interval_minutes = interval + settings.save() + return 0 + + if args.action == "uninstall": + if plat == "macos": + from tagsync import platform_macos + removed = platform_macos.uninstall_schedule() + elif plat == "windows": + from tagsync import platform_windows + removed = platform_windows.uninstall_schedule() + else: + print(f"Unsupported platform: {plat}", file=sys.stderr) + return 2 + print("Schedule removed." if removed else "No schedule was installed.") + settings.schedule_installed = False + settings.save() + return 0 + + if args.action == "status": + sched = _schedule_status(plat) + for key, val in sched.items(): + print(f"{key}: {val}") + return 0 + + return 2 + + +def cmd_reset(args: argparse.Namespace) -> int: + if args.token or args.all: + config.clear_token() + print("Token cleared.") + if args.settings or args.all: + if config.CONFIG_FILE.exists(): + config.CONFIG_FILE.unlink() + print("Settings cleared.") + if args.state or args.all: + if config.STATE_DB_FILE.exists(): + config.STATE_DB_FILE.unlink() + print("State DB cleared (next run will re-examine every file).") + if not (args.token or args.settings or args.state or args.all): + print("Nothing to do. Use --token, --settings, --state, or --all.") + return 2 + return 0 + + +def _schedule_status(plat: str) -> dict: + try: + if plat == "macos": + from tagsync import platform_macos + return platform_macos.schedule_status() + if plat == "windows": + from tagsync import platform_windows + return platform_windows.schedule_status() + except Exception as e: + return {"error": str(e)} + return {"error": f"unsupported platform: {plat}"} + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="tagsync", + description="Sync Dropbox web tags to local OS-native tags.", + ) + sub = p.add_subparsers(dest="cmd", required=True) + + w = sub.add_parser("wizard", help="Interactive setup (run this first)") + w.set_defaults(func=cmd_wizard) + + r = sub.add_parser("run", help="Perform a one-shot sync") + r.add_argument("--dry-run", action="store_true", + help="Report what would happen without writing tags") + r.add_argument("--verbose", action="store_true") + r.set_defaults(func=cmd_run) + + s = sub.add_parser("status", help="Show current configuration and schedule state") + s.set_defaults(func=cmd_status) + + sch = sub.add_parser("schedule", help="Manage the background scheduled task") + sch.add_argument("action", choices=["install", "uninstall", "status"]) + sch.add_argument("--interval", type=int, default=0, + help="Minutes between runs (install only, default 60)") + sch.set_defaults(func=cmd_schedule) + + rs = sub.add_parser("reset", help="Clear stored token, settings, or state") + rs.add_argument("--token", action="store_true") + rs.add_argument("--settings", action="store_true") + rs.add_argument("--state", action="store_true") + rs.add_argument("--all", action="store_true") + rs.set_defaults(func=cmd_reset) + + return p + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tagsync/__init__.py b/tagsync/__init__.py new file mode 100644 index 0000000..f923315 --- /dev/null +++ b/tagsync/__init__.py @@ -0,0 +1,3 @@ +"""Dropbox tag sync: mirror Dropbox web tags to local OS-native tags.""" + +__version__ = "0.1.0" diff --git a/tagsync/config.py b/tagsync/config.py new file mode 100644 index 0000000..69be8f0 --- /dev/null +++ b/tagsync/config.py @@ -0,0 +1,94 @@ +"""Settings file + secure token storage.""" +from __future__ import annotations + +import json +import os +import stat +from dataclasses import asdict, dataclass, field +from pathlib import Path + +CONFIG_DIR = Path.home() / ".dropbox_tag_sync" +CONFIG_FILE = CONFIG_DIR / "config.json" +STATE_DB_FILE = CONFIG_DIR / "state.db" +TOKEN_FALLBACK_FILE = CONFIG_DIR / "token" + +KEYRING_SERVICE = "dropbox_tag_sync" +KEYRING_USER = "default" + + +@dataclass +class Settings: + """Persisted per-user settings.""" + + local_root: str = "" + platform: str = "" # "macos" or "windows" + exiftool: str = "" # windows only + skip_online_only: bool = True # windows only + write_sidecars: bool = False # windows only + schedule_installed: bool = False + schedule_interval_minutes: int = 60 + + def save(self, path: Path = CONFIG_FILE) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(asdict(self), indent=2)) + + @classmethod + def load(cls, path: Path = CONFIG_FILE) -> "Settings": + if not path.exists(): + return cls() + data = json.loads(path.read_text()) + # Filter to known fields so old configs don't crash on unknown keys. + known = {f for f in cls.__dataclass_fields__} + return cls(**{k: v for k, v in data.items() if k in known}) + + +# --- Token storage ------------------------------------------------------------ +# +# Primary: OS keyring (macOS Keychain, Windows Credential Manager). +# Fallback: ~/.dropbox_tag_sync/token (mode 0600). Used only if keyring +# import fails for some reason. + + +def _keyring_available() -> bool: + try: + import keyring # noqa: F401 + return True + except ImportError: + return False + + +def save_token(token: str) -> str: + """Persist token. Returns a human-readable location description.""" + if _keyring_available(): + import keyring + keyring.set_password(KEYRING_SERVICE, KEYRING_USER, token) + return "system keyring" + TOKEN_FALLBACK_FILE.parent.mkdir(parents=True, exist_ok=True) + TOKEN_FALLBACK_FILE.write_text(token) + try: + os.chmod(TOKEN_FALLBACK_FILE, stat.S_IRUSR | stat.S_IWUSR) + except OSError: + pass + return f"plaintext file at {TOKEN_FALLBACK_FILE} (keyring unavailable)" + + +def load_token() -> str | None: + if _keyring_available(): + import keyring + tok = keyring.get_password(KEYRING_SERVICE, KEYRING_USER) + if tok: + return tok + if TOKEN_FALLBACK_FILE.exists(): + return TOKEN_FALLBACK_FILE.read_text().strip() or None + return None + + +def clear_token() -> None: + if _keyring_available(): + import keyring + try: + keyring.delete_password(KEYRING_SERVICE, KEYRING_USER) + except Exception: + pass + if TOKEN_FALLBACK_FILE.exists(): + TOKEN_FALLBACK_FILE.unlink() diff --git a/tagsync/core.py b/tagsync/core.py new file mode 100644 index 0000000..b4cf5b0 --- /dev/null +++ b/tagsync/core.py @@ -0,0 +1,267 @@ +"""Shared Dropbox listing, tag fetching, state DB, and sync loop. + +Platform-agnostic. The sync loop takes a `Writer` object that knows how to +write tags on the current OS (see `platform_macos.py` / `platform_windows.py`). +""" +from __future__ import annotations + +import logging +import os +import sqlite3 +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Callable, Iterator, Protocol + +if TYPE_CHECKING: + import dropbox + from dropbox.files import FileMetadata + +TAG_BATCH = 100 # Dropbox files/tags/get accepts up to 100 paths per call. +log = logging.getLogger(__name__) + + +# --- Data shapes -------------------------------------------------------------- + + +@dataclass +class TaggedFile: + path_display: str # case-preserving Dropbox path (e.g. /Photos/IMG.jpg) + path_lower: str # lowercase Dropbox path (stable key) + local_path: Path + tags: list[str] # sorted tag names + + +@dataclass +class SyncStats: + listed: int = 0 + applied: int = 0 + cleared: int = 0 + skipped: int = 0 + missing: int = 0 + unsupported: int = 0 + online_only: int = 0 + errors: int = 0 + + def summary(self) -> str: + return ( + f"applied={self.applied} cleared={self.cleared} " + f"skipped={self.skipped} missing={self.missing} " + f"unsupported={self.unsupported} online-only={self.online_only} " + f"errors={self.errors}" + ) + + +class Writer(Protocol): + """Platform-specific tag writer.""" + + def supports(self, path: Path) -> bool: ... + def is_online_only(self, path: Path) -> bool: ... + def write_tags(self, path: Path, tags: list[str]) -> None: ... + + +# --- Dropbox auth + listing --------------------------------------------------- + + +def _dropbox_client(token: str): + import dropbox + return dropbox.Dropbox(token) + + +def verify_token(token: str) -> dict: + """Validate a token and return account info. Raises on failure.""" + dbx = _dropbox_client(token) + account = dbx.users_get_current_account() + return { + "name": account.name.display_name, + "email": account.email, + "account_id": account.account_id, + } + + +def _list_all_files(dbx) -> list: + from dropbox.files import FileMetadata + files: list = [] + result = dbx.files_list_folder("", recursive=True, include_deleted=False) + files.extend(e for e in result.entries if isinstance(e, FileMetadata)) + while result.has_more: + result = dbx.files_list_folder_continue(result.cursor) + files.extend(e for e in result.entries if isinstance(e, FileMetadata)) + return files + + +def _fetch_tags(dbx, files: list) -> Iterator[tuple]: + by_path = {f.path_lower: f for f in files} + paths = list(by_path.keys()) + for i in range(0, len(paths), TAG_BATCH): + batch = paths[i : i + TAG_BATCH] + resp = dbx.files_tags_get(batch) + for entry in resp.paths_to_tags: + tag_names = sorted( + t.get_user_generated_tag().tag_text for t in entry.tags + ) + meta = by_path.get(entry.path) + if meta is not None: + yield meta, tag_names + + +def _local_path(path_display: str, local_root: Path) -> Path: + return local_root / path_display.lstrip("/").replace("/", os.sep) + + +def collect_tagged_files( + token: str, + local_root: Path, + on_list_progress: Callable[[int], None] | None = None, + on_tag_progress: Callable[[int, int], None] | None = None, +) -> list[TaggedFile]: + """Return every Dropbox file that currently has at least one tag. + + Used by the wizard preview step. Does not include untagged files. + """ + dbx = _dropbox_client(token) + files = _list_all_files(dbx) + if on_list_progress: + on_list_progress(len(files)) + tagged: list[TaggedFile] = [] + total = len(files) + done = 0 + for meta, tags in _fetch_tags(dbx, files): + done += 1 + if on_tag_progress and done % TAG_BATCH == 0: + on_tag_progress(done, total) + if tags: + tagged.append( + TaggedFile( + path_display=meta.path_display, + path_lower=meta.path_lower, + local_path=_local_path(meta.path_display, local_root), + tags=tags, + ) + ) + if on_tag_progress: + on_tag_progress(done, total) + return tagged + + +# --- State DB ----------------------------------------------------------------- + + +def open_state(db_path: Path) -> sqlite3.Connection: + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(db_path) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS file_tags ( + path_lower TEXT PRIMARY KEY, + tags_csv TEXT NOT NULL, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + conn.commit() + return conn + + +# --- Main sync loop ----------------------------------------------------------- + + +def run_sync( + token: str, + local_root: Path, + state_db: Path, + writer: Writer, + *, + skip_online_only: bool = False, + dry_run: bool = False, + on_progress: Callable[[TaggedFile, str], None] | None = None, +) -> SyncStats: + """Full one-shot sync: pull tags from Dropbox, write to local files. + + `on_progress(tagged_file, action)` is called for each file that is + actually touched. `action` is one of: 'applied', 'cleared', 'skipped', + 'missing', 'unsupported', 'online', 'error'. + """ + dbx = _dropbox_client(token) + state = open_state(state_db) + stats = SyncStats() + + files = _list_all_files(dbx) + stats.listed = len(files) + + for meta, tags in _fetch_tags(dbx, files): + local = _local_path(meta.path_display, local_root) + tags_csv = ",".join(tags) + tf = TaggedFile( + path_display=meta.path_display, + path_lower=meta.path_lower, + local_path=local, + tags=tags, + ) + + row = state.execute( + "SELECT tags_csv FROM file_tags WHERE path_lower = ?", + (meta.path_lower,), + ).fetchone() + + if row and row[0] == tags_csv: + stats.skipped += 1 + continue + + if not tags and not row: + # Untagged file we've never tagged — nothing to do. + stats.skipped += 1 + continue + + if not local.exists(): + log.warning("not found locally: %s", local) + stats.missing += 1 + if on_progress: + on_progress(tf, "missing") + continue + + if skip_online_only and writer.is_online_only(local): + stats.online_only += 1 + if on_progress: + on_progress(tf, "online") + continue + + if not writer.supports(local): + stats.unsupported += 1 + if on_progress: + on_progress(tf, "unsupported") + continue + + if dry_run: + if tags: + stats.applied += 1 + else: + stats.cleared += 1 + if on_progress: + on_progress(tf, "applied" if tags else "cleared") + continue + + try: + writer.write_tags(local, tags) + state.execute( + "INSERT INTO file_tags (path_lower, tags_csv) VALUES (?, ?) " + "ON CONFLICT(path_lower) DO UPDATE SET " + " tags_csv = excluded.tags_csv, " + " updated_at = CURRENT_TIMESTAMP", + (meta.path_lower, tags_csv), + ) + state.commit() + if tags: + stats.applied += 1 + if on_progress: + on_progress(tf, "applied") + else: + stats.cleared += 1 + if on_progress: + on_progress(tf, "cleared") + except Exception as e: + log.error("tagging failed on %s: %s", local, e) + stats.errors += 1 + if on_progress: + on_progress(tf, "error") + + return stats diff --git a/tagsync/detect.py b/tagsync/detect.py new file mode 100644 index 0000000..dcab722 --- /dev/null +++ b/tagsync/detect.py @@ -0,0 +1,95 @@ +"""Environment detection: platform, Dropbox folder, dependencies.""" +from __future__ import annotations + +import importlib.util +import json +import os +import platform +import shutil +import sys +from pathlib import Path +from typing import Literal + +Platform = Literal["macos", "windows", "linux"] + + +def detect_platform() -> Platform: + """Return 'macos', 'windows', or 'linux'.""" + system = platform.system() + if system == "Darwin": + return "macos" + if system == "Windows": + return "windows" + return "linux" + + +def platform_label() -> str: + """Human-friendly OS label, e.g. 'macOS 14.4'.""" + system = platform.system() + if system == "Darwin": + return f"macOS {platform.mac_ver()[0]}" + if system == "Windows": + return f"Windows {platform.release()}" + return f"{system} {platform.release()}" + + +def python_label() -> str: + return f"Python {sys.version.split()[0]}" + + +def _dropbox_info_candidates() -> list[Path]: + """Candidate paths for Dropbox's info.json config file.""" + home = Path.home() + paths: list[Path] = [] + if sys.platform == "win32": + for env_var in ("APPDATA", "LOCALAPPDATA"): + val = os.environ.get(env_var) + if val: + paths.append(Path(val) / "Dropbox" / "info.json") + else: + paths.append(home / ".dropbox" / "info.json") + paths.append( + home / "Library" / "Application Support" / "Dropbox" / "info.json" + ) + return paths + + +def detect_dropbox_folder() -> Path | None: + """Find the local Dropbox sync folder. + + First tries Dropbox's own info.json (authoritative), then falls back to + common default locations. + """ + for info_path in _dropbox_info_candidates(): + if not info_path.exists(): + continue + try: + info = json.loads(info_path.read_text()) + except (OSError, json.JSONDecodeError): + continue + for account_type in ("personal", "business"): + account = info.get(account_type) + if account and "path" in account: + candidate = Path(account["path"]) + if candidate.is_dir(): + return candidate + + home = Path.home() + for candidate in (home / "Dropbox", home / "Dropbox (Personal)"): + if candidate.is_dir(): + return candidate + return None + + +def has_module(module_name: str) -> bool: + """True if a Python module is importable without actually importing it.""" + return importlib.util.find_spec(module_name) is not None + + +def find_exiftool() -> Path | None: + """Locate exiftool on PATH, if present.""" + for name in ("exiftool", "exiftool.exe"): + found = shutil.which(name) + if found: + return Path(found) + return None diff --git a/tagsync/platform_macos.py b/tagsync/platform_macos.py new file mode 100644 index 0000000..72a860d --- /dev/null +++ b/tagsync/platform_macos.py @@ -0,0 +1,125 @@ +"""macOS tag writer (Finder xattr) + launchd scheduler.""" +from __future__ import annotations + +import errno +import os +import plistlib +import subprocess +import sys +from pathlib import Path +from textwrap import dedent + +FINDER_TAGS_XATTR = "com.apple.metadata:_kMDItemUserTags" +LAUNCHD_LABEL = "com.dropboxtagsync.agent" +LAUNCHD_PLIST = Path.home() / "Library" / "LaunchAgents" / f"{LAUNCHD_LABEL}.plist" +LOG_PATH = Path.home() / "Library" / "Logs" / "dropbox-tag-sync.log" + +# Errno returned by macOS when an xattr is missing (ENOATTR = 93). +_NO_ATTR_ERRNOS = {93, errno.ENODATA if hasattr(errno, "ENODATA") else 61} + + +class MacWriter: + """Writer that stores tags in the Finder xattr. All file types supported.""" + + def supports(self, path: Path) -> bool: + return True + + def is_online_only(self, path: Path) -> bool: + # macOS Dropbox doesn't use a public placeholder attribute equivalent + # to Windows. If this becomes a concern we can add detection via + # Dropbox's File Provider APIs, but for now assume files are materialized. + return False + + def write_tags(self, path: Path, tags: list[str]) -> None: + if not tags: + try: + os.removexattr(path, FINDER_TAGS_XATTR) + except FileNotFoundError: + pass + except OSError as e: + if e.errno not in _NO_ATTR_ERRNOS: + raise + return + plist = plistlib.dumps(tags, fmt=plistlib.FMT_BINARY) + os.setxattr(path, FINDER_TAGS_XATTR, plist) + + +def install_schedule( + entry_script: Path, + interval_minutes: int = 60, + python_exe: str | None = None, +) -> Path: + """Create and load a launchd agent that runs the sync every N minutes.""" + python_exe = python_exe or sys.executable + interval_seconds = max(60, interval_minutes * 60) + + plist_body = dedent( + f"""\ + + + + Label{LAUNCHD_LABEL} + ProgramArguments + {python_exe} + {entry_script} + run + + StartInterval{interval_seconds} + RunAtLoad + StandardOutPath{LOG_PATH} + StandardErrorPath{LOG_PATH} + + """ + ) + + LAUNCHD_PLIST.parent.mkdir(parents=True, exist_ok=True) + LOG_PATH.parent.mkdir(parents=True, exist_ok=True) + LAUNCHD_PLIST.write_text(plist_body) + + # Unload any prior version, then load the new one. + subprocess.run( + ["launchctl", "unload", str(LAUNCHD_PLIST)], + capture_output=True, + text=True, + ) + result = subprocess.run( + ["launchctl", "load", str(LAUNCHD_PLIST)], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"launchctl load failed: {result.stderr.strip()}") + return LAUNCHD_PLIST + + +def uninstall_schedule() -> bool: + """Remove the launchd agent. Returns True if it existed.""" + if not LAUNCHD_PLIST.exists(): + return False + subprocess.run( + ["launchctl", "unload", str(LAUNCHD_PLIST)], + capture_output=True, + text=True, + ) + LAUNCHD_PLIST.unlink() + return True + + +def schedule_status() -> dict: + """Return info about whether the agent is installed/loaded.""" + installed = LAUNCHD_PLIST.exists() + loaded = False + if installed: + result = subprocess.run( + ["launchctl", "list", LAUNCHD_LABEL], + capture_output=True, + text=True, + ) + loaded = result.returncode == 0 + return { + "installed": installed, + "loaded": loaded, + "plist": str(LAUNCHD_PLIST), + "log": str(LOG_PATH), + } diff --git a/tagsync/platform_windows.py b/tagsync/platform_windows.py new file mode 100644 index 0000000..29b5629 --- /dev/null +++ b/tagsync/platform_windows.py @@ -0,0 +1,129 @@ +"""Windows tag writer (XMP via exiftool) + Task Scheduler.""" +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +TASK_NAME = "DropboxTagSync" + +# Windows file attribute flags identifying cloud placeholders. +FILE_ATTRIBUTE_RECALL_ON_DATA_ACCESS = 0x00400000 +FILE_ATTRIBUTE_OFFLINE = 0x00001000 + +# Conservative list of extensions exiftool can reliably write XMP into. +# Full list: https://exiftool.org/#supported (see "Write" column). +XMP_WRITABLE_EXTS: set[str] = { + # Images + ".jpg", ".jpeg", ".jpe", ".jfif", ".png", ".tif", ".tiff", ".gif", + ".heic", ".heif", ".webp", ".psd", ".dng", ".cr2", ".cr3", ".nef", + ".arw", ".orf", ".raf", ".rw2", ".svg", + # Documents + ".pdf", ".docx", ".xlsx", ".pptx", ".odt", ".ods", ".odp", ".epub", + # Audio / video + ".mp3", ".m4a", ".wav", ".flac", ".aac", ".ogg", + ".mp4", ".m4v", ".mov", ".avi", ".mkv", +} + + +class WindowsWriter: + """Writes Explorer 'Tags' via XMP:Subject using exiftool. + + Can optionally write a .tags.json sidecar for unsupported file types. + """ + + def __init__(self, exiftool: str, write_sidecars: bool = False) -> None: + self.exiftool = exiftool + self.write_sidecars = write_sidecars + + def supports(self, path: Path) -> bool: + if path.suffix.lower() in XMP_WRITABLE_EXTS: + return True + return self.write_sidecars + + def is_online_only(self, path: Path) -> bool: + try: + attrs = path.stat().st_file_attributes # type: ignore[attr-defined] + except (AttributeError, OSError): + return False + return bool( + attrs & (FILE_ATTRIBUTE_RECALL_ON_DATA_ACCESS | FILE_ATTRIBUTE_OFFLINE) + ) + + def write_tags(self, path: Path, tags: list[str]) -> None: + if path.suffix.lower() in XMP_WRITABLE_EXTS: + self._write_xmp(path, tags) + elif self.write_sidecars: + self._write_sidecar(path, tags) + else: + raise RuntimeError(f"unsupported file type for tagging: {path.suffix}") + + def _write_xmp(self, path: Path, tags: list[str]) -> None: + args = [self.exiftool, "-overwrite_original", "-q", "-XMP-dc:Subject="] + for tag in tags: + args.append(f"-XMP-dc:Subject+={tag}") + args.append(str(path)) + result = subprocess.run(args, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError(f"exiftool failed: {result.stderr.strip()}") + + @staticmethod + def _write_sidecar(path: Path, tags: list[str]) -> None: + sidecar = path.with_suffix(path.suffix + ".tags.json") + if tags: + sidecar.write_text(json.dumps({"tags": tags}, indent=2)) + else: + try: + sidecar.unlink() + except FileNotFoundError: + pass + + +def install_schedule( + entry_script: Path, + interval_minutes: int = 60, + python_exe: str | None = None, +) -> None: + """Create a recurring Windows scheduled task.""" + python_exe = python_exe or sys.executable + # Command is quoted as a single string since /tr takes one argument. + task_cmd = f'"{python_exe}" "{entry_script}" run' + subprocess.run( + [ + "schtasks", "/create", + "/tn", TASK_NAME, + "/tr", task_cmd, + "/sc", "minute", + "/mo", str(max(1, interval_minutes)), + "/rl", "limited", + "/f", + ], + check=True, + ) + + +def uninstall_schedule() -> bool: + """Delete the scheduled task. Returns True if it existed.""" + result = subprocess.run( + ["schtasks", "/delete", "/tn", TASK_NAME, "/f"], + capture_output=True, + text=True, + ) + return result.returncode == 0 + + +def schedule_status() -> dict: + """Query Task Scheduler for this task's status.""" + result = subprocess.run( + ["schtasks", "/query", "/tn", TASK_NAME, "/fo", "LIST"], + capture_output=True, + text=True, + ) + installed = result.returncode == 0 + return { + "installed": installed, + "loaded": installed, + "task_name": TASK_NAME, + "details": result.stdout.strip() if installed else "", + } diff --git a/tagsync/wizard.py b/tagsync/wizard.py new file mode 100644 index 0000000..3cd5958 --- /dev/null +++ b/tagsync/wizard.py @@ -0,0 +1,455 @@ +"""Interactive setup wizard. + +Runs a step-by-step flow: detect OS, locate Dropbox, collect & verify a +token, preview tagged files, apply tags, and optionally install a +recurring schedule. Safe to re-run — existing settings are offered as +defaults. +""" +from __future__ import annotations + +import getpass +import sys +import textwrap +import time +import webbrowser +from pathlib import Path + +from . import config, core, detect + + +# --- Small UI helpers --------------------------------------------------------- +# +# Kept inline (no extra module) so the wizard stays readable in one place. + +_USE_COLOR = sys.stdout.isatty() + + +def _c(s: str, code: str) -> str: + return f"\033[{code}m{s}\033[0m" if _USE_COLOR else s + + +def _bold(s: str) -> str: return _c(s, "1") +def _dim(s: str) -> str: return _c(s, "2") +def _green(s: str) -> str: return _c(s, "32") +def _yellow(s: str) -> str: return _c(s, "33") +def _red(s: str) -> str: return _c(s, "31") +def _blue(s: str) -> str: return _c(s, "34") + + +def header(title: str) -> None: + bar = "=" * max(len(title), 40) + print() + print(_bold(title)) + print(_bold(bar)) + + +def step(n: str, title: str) -> None: + print() + print(_blue(f"[{n}] ") + _bold(title)) + + +def info(msg: str) -> None: + print(f" {msg}") + + +def dim(msg: str) -> None: + print(f" {_dim(msg)}") + + +def success(msg: str) -> None: + print(f" {_green('OK')} {msg}") + + +def warn(msg: str) -> None: + print(f" {_yellow('!')} {msg}") + + +def error(msg: str) -> None: + print(f" {_red('x')} {msg}", file=sys.stderr) + + +def paragraph(text: str) -> None: + for line in textwrap.dedent(text).strip().splitlines(): + print(f" {line}") + + +def prompt_yesno(question: str, default: bool = True) -> bool: + suffix = "[Y/n]" if default else "[y/N]" + while True: + raw = input(f" {question} {suffix}: ").strip().lower() + if raw == "": + return default + if raw in ("y", "yes"): + return True + if raw in ("n", "no"): + return False + + +def prompt_text(question: str, default: str = "") -> str: + suffix = f" [{default}]" if default else "" + raw = input(f" {question}{suffix}: ").strip() + return raw or default + + +def prompt_int(question: str, default: int) -> int: + while True: + raw = prompt_text(question, str(default)) + try: + return int(raw) + except ValueError: + error("Please enter a number.") + + +# --- Wizard steps ------------------------------------------------------------- + + +def run_wizard(entry_script: Path) -> int: + """Entry point for `tagsync.py wizard`. Returns process exit code.""" + header("Dropbox Tag Sync — Setup Wizard") + + plat = detect.detect_platform() + info(f"System: {detect.platform_label()}") + info(f"Runtime: {detect.python_label()}") + + if plat == "linux": + print() + error("This tool currently supports macOS and Windows only.") + return 1 + + existing = config.Settings.load() + + local_root = _step_dropbox_folder(existing) + if local_root is None: + return 1 + + if not _step_check_deps(plat): + return 1 + + exiftool_path = "" + if plat == "windows": + tool = detect.find_exiftool() + if tool is None: + error("ExifTool is required on Windows to write file tags.") + info("Download it here: https://exiftool.org/") + if prompt_yesno("Open the ExifTool download page now?", default=True): + webbrowser.open("https://exiftool.org/") + info("After installing, re-run the wizard.") + return 1 + exiftool_path = str(tool) + success(f"ExifTool found: {exiftool_path}") + + token = _step_token(existing_token=config.load_token()) + if token is None: + return 1 + + skip_online_only = existing.skip_online_only + write_sidecars = existing.write_sidecars + if plat == "windows": + skip_online_only, write_sidecars = _step_windows_options( + skip_online_only, write_sidecars + ) + + # Save settings so far — even if the user aborts the sync step, they can + # come back and run `tagsync.py run` without re-entering everything. + settings = config.Settings( + local_root=str(local_root), + platform=plat, + exiftool=exiftool_path, + skip_online_only=skip_online_only, + write_sidecars=write_sidecars, + schedule_installed=existing.schedule_installed, + schedule_interval_minutes=existing.schedule_interval_minutes, + ) + settings.save() + + tagged = _step_preview(token, local_root) + if tagged is None: + return 1 + + if tagged: + applied = _step_apply( + token, local_root, settings, entry_script, tagged + ) + else: + info("There's nothing to tag right now. You can always re-run later.") + applied = True + + _step_schedule(entry_script, settings) + + print() + success("Setup complete.") + info( + "Re-run anytime with: " + + _bold(f"{sys.executable} {entry_script} run") + ) + info( + "Check schedule with: " + + _bold(f"{sys.executable} {entry_script} status") + ) + return 0 if applied else 1 + + +# --- Step: locate Dropbox folder --------------------------------------------- + + +def _step_dropbox_folder(existing: config.Settings) -> Path | None: + step("1/6", "Locate your Dropbox folder") + + candidate = None + if existing.local_root and Path(existing.local_root).is_dir(): + candidate = Path(existing.local_root) + info(f"Previously configured: {candidate}") + else: + candidate = detect.detect_dropbox_folder() + if candidate: + info(f"Auto-detected: {candidate}") + else: + warn("Could not auto-detect your Dropbox folder.") + + if candidate and prompt_yesno("Is this the right folder?", default=True): + return candidate + + while True: + typed = prompt_text("Enter the full path to your Dropbox folder") + if not typed: + error("Path is required.") + continue + path = Path(typed).expanduser() + if not path.is_dir(): + error(f"Not a directory: {path}") + continue + return path + + +# --- Step: dependency check -------------------------------------------------- + + +def _step_check_deps(plat: str) -> bool: + step("2/6", "Check dependencies") + + ok = True + if detect.has_module("dropbox"): + success("dropbox package installed") + else: + error("The 'dropbox' Python package is not installed.") + info("Install it with: pip install -r requirements.txt") + ok = False + + if detect.has_module("keyring"): + success("keyring installed (secure token storage)") + else: + warn("keyring not installed — token will fall back to a plaintext file.") + info("For best security: pip install keyring") + return ok + + +# --- Step: token ------------------------------------------------------------- + + +_TOKEN_INSTRUCTIONS = """\ +A Dropbox access token is needed so this tool can read your tags. +The token only has permission to READ METADATA — it cannot modify or +delete anything in your Dropbox. + +You'll create the token on the Dropbox developer page: + 1. Click "Create app" + 2. Choose: Scoped access, Full Dropbox, name it "tag-sync" + 3. Open the Permissions tab, enable files.metadata.read, click Submit + 4. Return to the Settings tab, scroll to "Generated access token", + and click Generate + 5. Copy the token shown +""" + + +def _step_token(existing_token: str | None) -> str | None: + step("3/6", "Dropbox access token") + + token = None + if existing_token: + info("A saved token was found.") + if prompt_yesno("Use the existing token?", default=True): + token = existing_token + + if not token: + paragraph(_TOKEN_INSTRUCTIONS) + if prompt_yesno("Open the Dropbox developer page in your browser?", default=True): + webbrowser.open("https://www.dropbox.com/developers/apps") + token = getpass.getpass(" Paste your access token (input hidden): ").strip() + if not token: + error("No token provided.") + return None + + info("Verifying token with Dropbox...") + try: + account = core.verify_token(token) + except Exception as e: + error(f"Token verification failed: {e}") + return None + success(f"Signed in as {account['name']} <{account['email']}>") + + if token != existing_token: + location = config.save_token(token) + info(f"Token stored in {location}.") + return token + + +# --- Step: Windows-only behavior options ------------------------------------- + + +def _step_windows_options(skip_online_only: bool, write_sidecars: bool) -> tuple[bool, bool]: + step("3b/6", "Windows tagging options") + paragraph( + """ + Windows embeds tags inside files, so only certain formats can be + tagged (images, PDFs, Office docs, most media). A few options: + """ + ) + skip_online_only = prompt_yesno( + "Skip files that are online-only (not yet downloaded)?", + default=skip_online_only, + ) + write_sidecars = prompt_yesno( + "Also write .tags.json sidecar files for unsupported types? " + "(Not indexed by Explorer; useful for custom tooling)", + default=write_sidecars, + ) + return skip_online_only, write_sidecars + + +# --- Step: preview ----------------------------------------------------------- + + +def _step_preview(token: str, local_root: Path) -> list[core.TaggedFile] | None: + step("4/6", "Scan Dropbox for tagged files") + info("This may take a minute for large accounts...") + + start = time.time() + listed = {"n": 0} + last_tag_report = {"t": 0.0} + + def on_list(n: int) -> None: + listed["n"] = n + info(f" listed {n} files...") + + def on_tag(done: int, total: int) -> None: + now = time.time() + if now - last_tag_report["t"] < 1.0 and done < total: + return + last_tag_report["t"] = now + pct = int(100 * done / total) if total else 100 + info(f" fetched tags for {done}/{total} files ({pct}%)") + + try: + tagged = core.collect_tagged_files( + token, local_root, on_list_progress=on_list, on_tag_progress=on_tag + ) + except Exception as e: + error(f"Scan failed: {e}") + return None + + elapsed = time.time() - start + success( + f"Found {len(tagged)} tagged file(s) across {listed['n']} " + f"total ({elapsed:.1f}s)." + ) + + if tagged: + info("") + info("Sample:") + for tf in tagged[:5]: + dim(f" {tf.path_display} -> {', '.join(tf.tags)}") + if len(tagged) > 5: + dim(f" ... and {len(tagged) - 5} more") + return tagged + + +# --- Step: apply ------------------------------------------------------------- + + +def _step_apply( + token: str, + local_root: Path, + settings: config.Settings, + entry_script: Path, + tagged: list[core.TaggedFile], +) -> bool: + step("5/6", "Apply tags to local files") + + if settings.platform == "windows": + warn( + "On Windows, each tagged file will be modified in place and " + "Dropbox will re-upload it once." + ) + if not prompt_yesno(f"Tag {len(tagged)} file(s) now?", default=True): + info("Skipped. You can run this later with:") + info(f" {sys.executable} {entry_script} run") + return True + + writer = _build_writer(settings) + stats = core.run_sync( + token=token, + local_root=local_root, + state_db=config.STATE_DB_FILE, + writer=writer, + skip_online_only=settings.skip_online_only, + dry_run=False, + ) + print() + info(f"Result: {stats.summary()}") + if stats.errors: + warn(f"{stats.errors} file(s) failed to tag — see above for details.") + return False + success("All tags applied successfully.") + return True + + +def _build_writer(settings: config.Settings): + if settings.platform == "macos": + from .platform_macos import MacWriter + return MacWriter() + if settings.platform == "windows": + from .platform_windows import WindowsWriter + return WindowsWriter( + exiftool=settings.exiftool, + write_sidecars=settings.write_sidecars, + ) + raise RuntimeError(f"unsupported platform: {settings.platform}") + + +# --- Step: schedule ---------------------------------------------------------- + + +def _step_schedule(entry_script: Path, settings: config.Settings) -> None: + step("6/6", "Automatic sync") + paragraph( + """ + New Dropbox tags only show up locally after this tool runs. You + can schedule it to run automatically in the background. + """ + ) + + if not prompt_yesno("Install a scheduled task to sync periodically?", default=True): + info("Skipped. You can still run the tool manually.") + return + + interval = prompt_int("Run every how many minutes?", settings.schedule_interval_minutes or 60) + interval = max(5, interval) + + try: + if settings.platform == "macos": + from . import platform_macos + plist = platform_macos.install_schedule(entry_script, interval_minutes=interval) + success(f"Installed launchd agent: {plist}") + else: + from . import platform_windows + platform_windows.install_schedule(entry_script, interval_minutes=interval) + success(f"Installed scheduled task: {platform_windows.TASK_NAME}") + except Exception as e: + error(f"Failed to install schedule: {e}") + info("You can install it manually later with:") + info(f" {sys.executable} {entry_script} schedule install") + return + + settings.schedule_installed = True + settings.schedule_interval_minutes = interval + settings.save() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f451f93 --- /dev/null +++ b/tests/conftest.py @@ -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}") + ) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..bbd2736 --- /dev/null +++ b/tests/test_config.py @@ -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() diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..f6746ea --- /dev/null +++ b/tests/test_core.py @@ -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" diff --git a/tests/test_detect.py b/tests/test_detect.py new file mode 100644 index 0000000..27a0f53 --- /dev/null +++ b/tests/test_detect.py @@ -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 diff --git a/tests/test_platform_macos.py b/tests/test_platform_macos.py new file mode 100644 index 0000000..3673962 --- /dev/null +++ b/tests/test_platform_macos.py @@ -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 diff --git a/tests/test_platform_windows.py b/tests/test_platform_windows.py new file mode 100644 index 0000000..c0e75e1 --- /dev/null +++ b/tests/test_platform_windows.py @@ -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