diff --git a/.coverage b/.coverage
index 6326bd9..d619a43 100644
Binary files a/.coverage and b/.coverage differ
diff --git a/.github/workflows/publish-testpypi.yml b/.github/workflows/publish-testpypi.yml
deleted file mode 100644
index 1548447..0000000
--- a/.github/workflows/publish-testpypi.yml
+++ /dev/null
@@ -1,48 +0,0 @@
-name: Publish to TestPyPI (Manual)
-
-# Manual workflow to publish to TestPyPI for testing before production release
-on:
- workflow_dispatch:
- inputs:
- confirm:
- description: 'Type "publish" to confirm TestPyPI upload'
- required: true
- type: string
-
-permissions:
- contents: read
- id-token: write
-
-jobs:
- build_and_publish:
- name: Build and publish to TestPyPI
- runs-on: ubuntu-latest
- # Only run if user typed "publish" to confirm
- if: github.event.inputs.confirm == 'publish'
- environment: testpypi
-
- steps:
- - name: Checkout repository
- uses: actions/checkout@v4
-
- - name: Set up Python
- uses: actions/setup-python@v5
- with:
- python-version: "3.x"
-
- - name: Install build tooling
- run: python -m pip install --upgrade pip build
-
- - name: Build distributions
- run: python -m build --sdist --wheel
-
- - name: Publish package distributions to TestPyPI
- uses: pypa/gh-action-pypi-publish@release/v1
- with:
- repository-url: https://test.pypi.org/legacy/
- skip-existing: true
-
- - name: Output installation command
- run: |
- VERSION=$(python -c "import tomllib; print(tomllib.load(open('pyproject.toml', 'rb'))['project']['version'])")
- echo "::notice::Package published to TestPyPI. Test with: pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple fmd_api==$VERSION"
diff --git a/README.md b/README.md
index c09188b..3f70132 100644
--- a/README.md
+++ b/README.md
@@ -2,6 +2,7 @@
[](https://github.com/devinslick/fmd_api/actions/workflows/test.yml)
[](https://codecov.io/gh/devinslick/fmd_api)
+[](https://pypi.org/project/fmd-api/)
Modern, async Python client for the open‑source FMD (Find My Device) server. It handles authentication, key management, encrypted data decryption, location/picture retrieval, and common device commands with safe, validated helpers.
@@ -12,11 +13,7 @@ Modern, async Python client for the open‑source FMD (Find My Device) server. I
```bash
pip install fmd_api
```
-- Pre‑release (Test PyPI):
- ```bash
- pip install --pre --index-url https://test.pypi.org/simple/ \
- --extra-index-url https://pypi.org/simple/ fmd_api
- ```
+
## Quickstart
@@ -32,7 +29,9 @@ async def main():
# Fetch most recent locations and decrypt the latest
blobs = await client.get_locations(num_to_get=1)
- loc = json.loads(client.decrypt_data_blob(blobs[0]))
+ # decrypt_data_blob() returns raw bytes — decode then parse JSON for clarity
+ decrypted = client.decrypt_data_blob(blobs[0])
+ loc = json.loads(decrypted.decode("utf-8"))
print(loc["lat"], loc["lon"], loc.get("accuracy"))
# Take a picture (validated helper)
@@ -102,7 +101,7 @@ Tips:
## What’s in the box
- `FmdClient` (primary API)
- - Auth and key retrieval (salt → Argon2id → access token → private key decrypt)
+ - Auth and key retrieval (salt → Argon2id → access token → private key retrieval and decryption)
- Decrypt blobs (RSA‑OAEP wrapped AES‑GCM)
- Fetch data: `get_locations`, `get_pictures`
- Export: `export_data_zip(out_path)` — client-side packaging of all locations/pictures into ZIP (mimics web UI, no server endpoint)
@@ -121,6 +120,13 @@ Tips:
- `await device.refresh()` → hydrate cached state
- `await device.get_location()` → parsed last location
- `await device.get_picture_blobs(n)` + `await device.decode_picture(blob)`
+ - `await device.get_picture_metadata(n)` -> returns only metadata dicts (if the server exposes them)
+
+ IMPORTANT (breaking change in v2.0.5): legacy compatibility wrappers were removed.
+ The following legacy methods were removed from the `Device` API: `fetch_pictures`,
+ `get_pictures`, `download_photo`, `get_picture`, `take_front_photo`, and `take_rear_photo`.
+ Update your code to use `get_picture_blobs()`, `decode_picture()`, `take_front_picture()`
+ and `take_rear_picture()` instead.
- Commands: `await device.play_sound()`, `await device.take_front_picture()`,
`await device.take_rear_picture()`, `await device.lock(message=None)`,
`await device.wipe(pin="YourSecurePIN", confirm=True)`
@@ -143,6 +149,35 @@ async def main():
asyncio.run(main())
```
+### Example: Inspect pictures metadata (when available)
+
+Use `get_picture_blobs()` to fetch the raw server responses (strings or dicts). If you want a
+strongly-typed list of picture metadata objects (where the server provides metadata as JSON
+objects), use `get_picture_metadata()`, which filters for dict entries and returns only those.
+
+```python
+from fmd_api import FmdClient, Device
+
+async def inspect_metadata():
+ client = await FmdClient.create("https://fmd.example.com", "alice", "secret")
+ device = Device(client, "alice")
+
+ # Raw values may be strings (base64 blobs) or dicts (metadata). Keep raw when you need
+ # to decode or handle both forms yourself.
+ raw = await device.get_picture_blobs(10)
+
+ # If you want only metadata entries returned by the server, use get_picture_metadata().
+ # This returns a list of dict-like metadata objects (e.g. id/date/filename) and filters
+ # out any raw string blobs.
+ metadata = await device.get_picture_metadata(10)
+ for m in metadata:
+ print(m.get("id"), m.get("date"))
+
+ await client.close()
+
+asyncio.run(inspect_metadata())
+```
+
## Testing
### Functional tests
diff --git a/coverage.xml b/coverage.xml
index bb20cdd..4df4cce 100644
--- a/coverage.xml
+++ b/coverage.xml
@@ -1,561 +1,707 @@
-
-
-
-
-
- C:\Users\Devin\Repos\fmd_api\fmd_api
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+ C:\Users\micro\Repos\fmd_api\fmd_api
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/docs/MIGRATE_FROM_V1.md b/docs/MIGRATE_FROM_V1.md
index 0283d1d..59f26ae 100644
--- a/docs/MIGRATE_FROM_V1.md
+++ b/docs/MIGRATE_FROM_V1.md
@@ -73,8 +73,8 @@ client = await FmdClient.create("https://fmd.example.com", "alice", "secret")
| V1 | V2 (FmdClient) | V2 (Device) | Notes |
|----|----------------|-------------|-------|
-| `await api.take_picture('back')` | `await client.take_picture('back')` | `await device.take_rear_picture()` | Device method preferred (old: take_rear_photo deprecated) |
-| `await api.take_picture('front')` | `await client.take_picture('front')` | `await device.take_front_picture()` | Device method preferred (old: take_front_photo deprecated) |
+| `await api.take_picture('back')` | `await client.take_picture('back')` | `await device.take_rear_picture()` | Device method preferred (legacy wrapper `take_rear_photo()` removed in v2.0.5) |
+| `await api.take_picture('front')` | `await client.take_picture('front')` | `await device.take_front_picture()` | Device method preferred (legacy wrapper `take_front_photo()` removed in v2.0.5) |
> Note: `Device.lock(message=None)` now supports passing an optional message string. The server may ignore the
> message if UI or server versions don't yet consume it, but the base lock command will still be executed.
@@ -92,8 +92,8 @@ client = await FmdClient.create("https://fmd.example.com", "alice", "secret")
| V1 | V2 (FmdClient) | V2 (Device) | Notes |
|----|----------------|-------------|-------|
-| `await api.get_pictures(10)` | `await client.get_pictures(10)` | `await device.get_picture_blobs(10)` | Both available (old: get_pictures/fetch_pictures deprecated) |
-| N/A | N/A | `await device.decode_picture(blob)` | Helper method (old: get_picture/download_photo deprecated) |
+| `await api.get_pictures(10)` | `await client.get_pictures(10)` | `await device.get_picture_blobs(10)` | Device wrapper methods `get_pictures()` / `fetch_pictures()` removed in v2.0.5 — use `get_picture_blobs()` |
+| N/A | N/A | `await device.decode_picture(blob)` | Helper method (legacy wrappers `get_picture()` / `download_photo()` removed in v2.0.5) |
### Export Data
@@ -153,8 +153,24 @@ await device.lock(message="Lost device") # Lock with message
await device.wipe(pin="YourSecurePIN", confirm=True) # Factory reset (DESTRUCTIVE, alphanumeric ASCII PIN + enabled setting)
# Pictures
+# V2: new client convenience methods provide both raw and metadata paths.
+#
+# - `get_picture_blobs(n)` returns raw server responses which may be either
+# - a string (usually a base64-encoded encrypted blob), or
+# - a dict/object containing metadata about the picture (id, date, filename, etc.)
+#
+# - `get_picture_metadata(n)` filters the raw values and returns only dict-like
+# metadata entries exposed by the server. Use this when you only need structured
+# metadata and don't want to handle raw encrypted blobs.
+#
+# Example: fetch raw blobs and decode the first picture
pictures = await device.get_picture_blobs(10)
photo_result = await device.decode_picture(pictures[0])
+
+# Example: fetch ONLY metadata entries (if server provides them)
+metadata_list = await device.get_picture_metadata(10)
+for meta in metadata_list:
+ print(f"Picture id={meta.get('id')}, date={meta.get('date')}, filename={meta.get('filename')}")
```
---
diff --git a/docs/release/v2.0.0_merge_request.md b/docs/release/v2.0.0_merge_request.md
index 1cc85f9..c3f37ef 100644
--- a/docs/release/v2.0.0_merge_request.md
+++ b/docs/release/v2.0.0_merge_request.md
@@ -24,8 +24,7 @@ This major release delivers a production-ready async client with robust TLS hand
- Typed package (`py.typed`) and mypy-clean.
- **CI/CD**
- GitHub Actions: lint (flake8), type-check (mypy), unit tests matrix (Ubuntu/Windows; Py 3.8–3.12).
- - Coverage with branch analysis and Codecov upload + badges.
- - Publish workflows for TestPyPI (non-main) and PyPI (main or Release).
+ - Coverage with branch analysis and Codecov upload + badges
## Breaking changes
- API surface moved to async:
@@ -65,7 +64,7 @@ This major release delivers a production-ready async client with robust TLS hand
- Coverage: ~83% overall; XML + branch coverage uploaded to Codecov (badges included).
- Workflows:
- `test.yml`: runs on push/PR for all branches (lint, mypy, unit tests matrix, coverage, optional functional tests).
- - `publish.yml`: builds on push/releases; publishes to TestPyPI for non-main pushes, PyPI on main or release.
+ - `publish.yml`: builds on push/releases; publishes to PyPI for main branch pushes and GitHub Releases. TestPyPI deployment is no longer used.
## Checklist
- [x] All unit tests pass
diff --git a/docs/release/v2.0.5.md b/docs/release/v2.0.5.md
new file mode 100644
index 0000000..9c10d57
--- /dev/null
+++ b/docs/release/v2.0.5.md
@@ -0,0 +1,53 @@
+# Release notes — v2.0.5
+
+Released: 2025-11-21
+
+This release focuses on two related improvements: full strict typing for the codebase
+and a clearer, more strongly-typed picture API. Important: this release removes several
+deprecated compatibility wrappers from the Device API — this is a breaking change.
+
+Highlights
+- Enforced strict typing across the entire repository
+ - `mypy` is now run in strict mode over the whole project (tests included).
+ - Many modules were updated with concrete, well-named type aliases and TypedDicts.
+ - New `fmd_api/types.py` centralizes JSON/response types and the new `PictureMetadata` TypedDict.
+ - Fixes were applied across the package to remove runtime Any usages and ensure
+ proper return annotations and runtime narrowing where needed.
+
+- Device picture APIs improved
+ - New method: `Device.get_picture_metadata(num_to_get: int = -1) -> List[PictureMetadata]`
+ - Returns only dict-like metadata entries (e.g., id, date, filename, size) when the
+ server exposes them.
+ - IMPORTANT: To remove historical ambiguity and simplify the public surface, this
+ release removes the old backward-compatible wrappers: `fetch_pictures()`,
+ `get_pictures()`, `download_photo()`, `get_picture()`, `take_front_photo()`,
+ and `take_rear_photo()` are no longer available. Calls must be migrated to the
+ canonical APIs: `get_picture_blobs()`, `decode_picture()`, `take_front_picture()`,
+ and `take_rear_picture()`.
+ - Tests updated and added to assert both behaviors (raw blobs preserved, metadata filtered).
+
+Why this change?
+- Strict typing improves long-term maintainability, reduces runtime bugs, and gives
+ contributors and users better IDE/typing support.
+- The new `get_picture_metadata` method makes it convenient to work with structured
+ metadata where available while preserving raw data semantics for older servers and
+ decoding use-cases.
+
+Notes for integrators (breaking change)
+- Removed APIs: `Device.fetch_pictures`, `Device.get_pictures`, `Device.download_photo`,
+ `Device.get_picture`, `Device.take_front_photo`, `Device.take_rear_photo` have been removed in v2.0.5.
+- Migration:
+ - Replace `fetch_pictures()` / `get_pictures()` -> `get_picture_blobs()`
+ - Replace `download_photo()` / `get_picture()` -> `decode_picture()`
+ - Replace `take_front_photo()` -> `take_front_picture()` and `take_rear_photo()` -> `take_rear_picture()`
+ - If your code depended on these wrappers for deprecated warning behavior, remove that handling.
+
+If you were relying on `get_picture_blobs` returning only dicts, use `get_picture_metadata()` to receive
+only metadata entries (dicts) where available.
+
+Migration
+- No breaking API changes. Prefer `get_picture_metadata` where you only need metadata.
+
+Acknowledgements
+- Thanks to the CI and test suite updates which made enforcing strict typing and landing
+ these changes safe across the codebase.
diff --git a/fmd_api/_version.py b/fmd_api/_version.py
index 34c5111..ff6ef86 100644
--- a/fmd_api/_version.py
+++ b/fmd_api/_version.py
@@ -1 +1 @@
-__version__ = "2.0.5"
+__version__ = "2.0.6"
diff --git a/fmd_api/client.py b/fmd_api/client.py
index 845fac3..ca73077 100644
--- a/fmd_api/client.py
+++ b/fmd_api/client.py
@@ -21,7 +21,8 @@
import logging
import time
import random
-from typing import Optional, List, Any, Dict, cast
+from typing import Any, Optional, List, Dict, cast, Union
+import ssl
from types import TracebackType
import aiohttp
@@ -33,6 +34,7 @@
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from .helpers import _pad_base64
+from .types import JSONType, AuthArtifacts
from .exceptions import FmdApiException
# Constants copied from original module to ensure parity
@@ -57,11 +59,11 @@ def __init__(
backoff_base: float = 0.5,
backoff_max: float = 10.0,
jitter: bool = True,
- ssl: Optional[Any] = None,
+ ssl: Optional[Union[bool, ssl.SSLContext]] = None,
conn_limit: Optional[int] = None,
conn_limit_per_host: Optional[int] = None,
keepalive_timeout: Optional[float] = None,
- ):
+ ) -> None:
# Enforce HTTPS only (FindMyDevice always uses TLS)
if base_url.lower().startswith("http://"):
raise ValueError("HTTPS is required for FmdClient base_url; plain HTTP is not allowed.")
@@ -115,7 +117,7 @@ async def create(
*,
cache_ttl: int = 30,
timeout: float = 30.0,
- ssl: Optional[Any] = None,
+ ssl: Optional[Union[bool, ssl.SSLContext]] = None,
conn_limit: Optional[int] = None,
conn_limit_per_host: Optional[int] = None,
keepalive_timeout: Optional[float] = None,
@@ -146,7 +148,8 @@ async def create(
async def _ensure_session(self) -> None:
if self._session is None or self._session.closed:
- connector_kwargs: Dict[str, Any] = {}
+ # Build TCPConnector with explicit arguments, only include values that are not None.
+ connector_kwargs: Dict[str, object] = {}
if self._ssl is not None:
connector_kwargs["ssl"] = self._ssl
if self._conn_limit is not None:
@@ -156,7 +159,10 @@ async def _ensure_session(self) -> None:
if self._keepalive_timeout is not None:
connector_kwargs["keepalive_timeout"] = self._keepalive_timeout
- connector = aiohttp.TCPConnector(**connector_kwargs)
+ # aiohttp's signature is fairly flexible; mypy can't easily verify **kwargs here.
+ # Use a cast to Any for the callsite so runtime behavior is unchanged but the
+ # type-checker won't complain about the dynamic set of keywords.
+ connector = aiohttp.TCPConnector(**cast(Any, connector_kwargs))
self._session = aiohttp.ClientSession(connector=connector)
async def close(self) -> None:
@@ -208,7 +214,11 @@ async def _get_salt(self, fmd_id: str) -> str:
return cast(str, await self._make_api_request("PUT", "/api/v1/salt", {"IDT": fmd_id, "Data": ""}))
async def _get_access_token(self, fmd_id: str, password_hash: str, session_duration: int) -> str:
- payload = {"IDT": fmd_id, "Data": password_hash, "SessionDurationSeconds": session_duration}
+ payload: Dict[str, JSONType] = {
+ "IDT": fmd_id,
+ "Data": password_hash,
+ "SessionDurationSeconds": session_duration,
+ }
return cast(str, await self._make_api_request("PUT", "/api/v1/requestAccess", payload))
async def _get_private_key_blob(self) -> str:
@@ -243,7 +253,7 @@ async def resume(
session_duration: int = 3600,
cache_ttl: int = 30,
timeout: float = 30.0,
- ssl: Optional[Any] = None,
+ ssl: Optional[Union[bool, ssl.SSLContext]] = None,
conn_limit: Optional[int] = None,
conn_limit_per_host: Optional[int] = None,
keepalive_timeout: Optional[float] = None,
@@ -278,7 +288,7 @@ async def resume(
inst.private_key = cast(RSAPrivateKey, serialization.load_der_private_key(pk_bytes, password=None))
return inst
- async def export_auth_artifacts(self) -> Dict[str, Any]:
+ async def export_auth_artifacts(self) -> AuthArtifacts:
"""Export current authentication artifacts for password-free resume."""
pk = self.private_key
if pk is None:
@@ -311,18 +321,44 @@ async def export_auth_artifacts(self) -> Dict[str, Any]:
}
@classmethod
- async def from_auth_artifacts(cls, artifacts: Dict[str, Any]) -> "FmdClient":
+ async def from_auth_artifacts(cls, artifacts: AuthArtifacts) -> "FmdClient":
required = ["base_url", "fmd_id", "access_token", "private_key"]
missing = [k for k in required if k not in artifacts]
if missing:
raise ValueError(f"Missing artifact fields: {missing}")
+ # Extract and validate required fields (TypedDict keys are optional so use .get()+assert)
+ # The TypedDict keys are optional, so fetch without indexing and perform runtime type checks
+ # Use Any-typed temporary variables so MyPy doesn't assume types from TypedDict keys
+ base_url_raw: Any = artifacts.get("base_url")
+ fmd_id_raw: Any = artifacts.get("fmd_id")
+ access_token_raw: Any = artifacts.get("access_token")
+ private_key_raw: Any = artifacts.get("private_key")
+
+ if (
+ not isinstance(base_url_raw, str)
+ or not isinstance(fmd_id_raw, str)
+ or not isinstance(access_token_raw, str)
+ or not isinstance(private_key_raw, str)
+ ):
+ raise ValueError("Missing or invalid artifact fields")
+
+ # Narrow to concrete types for resume() call
+ base_url = base_url_raw
+ fmd_id = fmd_id_raw
+ access_token = access_token_raw
+ private_key = private_key_raw
+ session_dur_raw: Any = artifacts.get("session_duration", 3600)
+ if not isinstance(session_dur_raw, int):
+ session_dur = 3600
+ else:
+ session_dur = session_dur_raw
return await cls.resume(
- artifacts["base_url"],
- artifacts["fmd_id"],
- artifacts["access_token"],
- artifacts["private_key"],
+ base_url,
+ fmd_id,
+ access_token,
+ private_key,
password_hash=artifacts.get("password_hash"),
- session_duration=artifacts.get("session_duration", 3600),
+ session_duration=session_dur,
)
async def drop_password(self) -> None:
@@ -381,13 +417,13 @@ async def _make_api_request(
self,
method: str,
endpoint: str,
- payload: Any,
+ payload: Dict[str, JSONType],
stream: bool = False,
expect_json: bool = True,
retry_auth: bool = True,
timeout: Optional[float] = None,
max_retries: Optional[int] = None,
- ) -> Any:
+ ) -> JSONType:
"""
Makes an API request and returns Data or text depending on expect_json/stream.
Mirrors get_all_locations/_make_api_request logic from original file (including 401 re-auth).
@@ -413,6 +449,7 @@ async def _make_api_request(
if self._password:
log.info("401 received: re-auth with raw password...")
await self.authenticate(self._fmd_id, self._password, self.session_duration)
+ # Narrow payload -> dict type ensured by signature
payload["IDT"] = self.access_token
return await self._make_api_request(
method,
@@ -486,13 +523,18 @@ async def _make_api_request(
if expect_json:
# server sometimes reports wrong content-type -> force JSON parse
try:
+ # Force JSON parse; narrow to dict/list cases to help static typing
json_data = await resp.json(content_type=None)
- # Sanitize: don't log full JSON which may contain tokens/sensitive data
- if log.isEnabledFor(logging.DEBUG):
- # Log safe metadata only
- keys = list(json_data.keys()) if isinstance(json_data, dict) else "non-dict"
- log.debug(f"{endpoint} JSON response received with keys: {keys}")
- return json_data["Data"]
+ # Narrow to dict responses (expected standard API envelope)
+ if isinstance(json_data, dict):
+ if log.isEnabledFor(logging.DEBUG):
+ # Log safe metadata only
+ keys = list(cast(Dict[str, JSONType], json_data).keys())
+ log.debug(f"{endpoint} JSON response received with keys: {keys}")
+ # Attempt to return the Data field; missing key will be handled below
+ return cast(JSONType, json_data["Data"])
+ # Non-dict JSON (list/primitive) -> fall back to text parsing
+ raise ValueError("JSON response not a dict; falling back to text")
except (KeyError, ValueError, json.JSONDecodeError) as e:
# fall back to text
log.debug(f"{endpoint} JSON parsing failed ({e}), trying as text")
@@ -508,8 +550,10 @@ async def _make_api_request(
log.debug(f"{endpoint} text response length: {len(text_data)}")
return text_data
else:
- # Return the aiohttp response for streaming consumers
- return resp
+ # Return the aiohttp response for streaming consumers (rare; keep runtime
+ # behavior but cast to JSONType for static typing so callers don't need
+ # to handle a separate return value in our codebase).
+ return cast(JSONType, resp)
except aiohttp.ClientConnectionError as e:
# Transient connection issues -> retry if allowed (avoid unsafe command repeats)
if attempts_left > 0 and not (is_command and method.upper() == "POST"):
@@ -540,6 +584,9 @@ async def get_locations(self, num_to_get: int = -1, skip_empty: bool = True, max
size_str = await self._make_api_request(
"PUT", "/api/v1/locationDataSize", {"IDT": self.access_token, "Data": ""}
)
+ # Narrow type: ensure the response can be converted to int (it should be a str or int)
+ if not isinstance(size_str, (str, int)):
+ raise FmdApiException(f"Unexpected value for locationDataSize: {size_str!r}")
size = int(size_str)
log.debug(f"Server reports {size} locations available")
if size == 0:
@@ -553,9 +600,15 @@ async def get_locations(self, num_to_get: int = -1, skip_empty: bool = True, max
for i in indices:
log.info(f" - Downloading location at index {i}...")
blob = await self._make_api_request(
- "PUT", "/api/v1/location", {"IDT": self.access_token, "Data": str(i)}
+ "PUT",
+ "/api/v1/location",
+ {"IDT": self.access_token, "Data": str(i)},
)
- locations.append(blob)
+ # Only accept string blobs for locations; defensive check to make mypy happy
+ if isinstance(blob, str):
+ locations.append(blob)
+ else:
+ log.warning("Skipping non-string location blob returned by server")
return locations
else:
num_to_download = min(num_to_get, size)
@@ -575,7 +628,9 @@ async def get_locations(self, num_to_get: int = -1, skip_empty: bool = True, max
for i in indices:
log.info(f" - Downloading location at index {i}...")
blob = await self._make_api_request("PUT", "/api/v1/location", {"IDT": self.access_token, "Data": str(i)})
- log.debug(f"Received blob type: {type(blob)}, length: {len(blob) if blob else 0}")
+ log.debug(
+ f"Received blob type: {type(blob)}, length: {len(blob) if isinstance(blob, (str, list, dict)) else 0}"
+ )
if blob and isinstance(blob, str) and blob.strip():
log.debug(f"First 100 chars: {blob[:100]}")
locations.append(blob)
@@ -583,14 +638,15 @@ async def get_locations(self, num_to_get: int = -1, skip_empty: bool = True, max
if len(locations) >= num_to_get and num_to_get != -1:
break
else:
- log.warning(f"Empty blob received for location index {i}, repr: {repr(blob[:50] if blob else blob)}")
+ sample = blob[:50] if isinstance(blob, (str, list, tuple, bytes)) else blob
+ log.warning(f"Empty blob received for location index {i}, repr: {repr(sample)}")
if not locations and num_to_get != -1:
log.warning(f"No valid locations found after checking " f"{min(max_attempts, size)} indices")
return locations
- async def get_pictures(self, num_to_get: int = -1, timeout: Optional[float] = None) -> List[Any]:
+ async def get_pictures(self, num_to_get: int = -1, timeout: Optional[float] = None) -> List[JSONType]:
"""Fetches all or the N most recent picture metadata blobs (raw server response)."""
req_timeout = aiohttp.ClientTimeout(total=timeout if timeout is not None else self.timeout)
try:
@@ -603,15 +659,22 @@ async def get_pictures(self, num_to_get: int = -1, timeout: Optional[float] = No
resp.raise_for_status()
json_data = await resp.json()
# Extract the Data field if it exists, otherwise use the response as-is
- all_pictures = json_data.get("Data", json_data) if isinstance(json_data, dict) else json_data
+ if isinstance(json_data, dict):
+ json_data_dict = cast(Dict[str, JSONType], json_data)
+ maybe: JSONType = json_data_dict.get("Data", json_data_dict)
+ else:
+ maybe = json_data
+
+ if not isinstance(maybe, list):
+ log.warning(f"Unexpected pictures response type: {type(maybe)}")
+ return []
+ all_pictures: List[JSONType] = maybe
except aiohttp.ClientError as e:
log.warning(f"Failed to get pictures: {e}. The endpoint may not exist or requires a different method.")
return []
- # Ensure all_pictures is a list
- if not isinstance(all_pictures, list):
- log.warning(f"Unexpected pictures response type: {type(all_pictures)}")
- return []
+ # all_pictures is assigned in the try/except above and validated to be a list
+ # (the prior checks already ensure we only reach here when it's a list).
if num_to_get == -1:
log.info(f"Found {len(all_pictures)} pictures to download.")
@@ -657,10 +720,10 @@ async def export_data_zip(self, out_path: str, include_pictures: bool = True) ->
# Fetch all locations
log.info("Fetching all locations...")
- location_blobs = await self.get_locations(num_to_get=-1, skip_empty=False)
+ location_blobs = cast(List[JSONType], await self.get_locations(num_to_get=-1, skip_empty=False))
# Fetch all pictures if requested
- picture_blobs = []
+ picture_blobs: List[JSONType] = []
if include_pictures:
log.info("Fetching all pictures...")
picture_blobs = await self.get_pictures(num_to_get=-1)
@@ -669,12 +732,16 @@ async def export_data_zip(self, out_path: str, include_pictures: bool = True) ->
log.info(f"Creating export ZIP at {out_path}...")
with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED) as zipf:
# Decrypt and add readable locations
- decrypted_locations = []
+ decrypted_locations: List[JSONType] = []
if location_blobs:
log.info(f"Decrypting {len(location_blobs)} locations...")
- for i, blob in enumerate(location_blobs):
+ for i, loc_blob in enumerate(location_blobs):
+ if not isinstance(loc_blob, str):
+ log.warning(f"Skipping non-text location blob at index {i}")
+ decrypted_locations.append({"error": "invalid blob type", "index": i})
+ continue
try:
- decrypted = self.decrypt_data_blob(blob)
+ decrypted = self.decrypt_data_blob(loc_blob)
loc_data = json.loads(decrypted)
decrypted_locations.append(loc_data)
except Exception as e:
@@ -682,12 +749,16 @@ async def export_data_zip(self, out_path: str, include_pictures: bool = True) ->
decrypted_locations.append({"error": str(e), "index": i})
# Decrypt and extract pictures as image files
- picture_file_list = []
+ picture_file_list: List[Dict[str, JSONType]] = []
if picture_blobs:
log.info(f"Decrypting and extracting {len(picture_blobs)} pictures...")
- for i, blob in enumerate(picture_blobs):
+ for i, pic_blob in enumerate(picture_blobs):
+ if not isinstance(pic_blob, str):
+ log.warning(f"Skipping non-text picture blob at index {i}")
+ picture_file_list.append({"index": i, "error": "invalid blob type"})
+ continue
try:
- decrypted = self.decrypt_data_blob(blob)
+ decrypted = self.decrypt_data_blob(pic_blob)
# Pictures are double-encoded: decrypt -> base64 string -> image bytes
inner_b64 = decrypted.decode("utf-8").strip()
from .helpers import b64_decode_padded
@@ -711,7 +782,7 @@ async def export_data_zip(self, out_path: str, include_pictures: bool = True) ->
picture_file_list.append({"index": i, "error": str(e)})
# Add metadata file (after processing so we have accurate counts)
- export_info = {
+ export_info: Dict[str, JSONType] = {
"export_date": datetime.now().isoformat(),
"fmd_id": self._fmd_id,
"location_count": len(location_blobs),
diff --git a/fmd_api/device.py b/fmd_api/device.py
index 15094da..210c525 100644
--- a/fmd_api/device.py
+++ b/fmd_api/device.py
@@ -8,8 +8,8 @@
import json
from datetime import datetime, timezone
-import warnings
-from typing import Optional, AsyncIterator, List, Dict, Any
+from typing import Optional, AsyncIterator, List, Dict, Union, cast
+from .types import JSONType, PictureMetadata
from .models import Location, PhotoResult
from .exceptions import OperationError
@@ -25,10 +25,10 @@ def _parse_location_blob(blob_b64: str) -> Location:
class Device:
- def __init__(self, client: FmdClient, fmd_id: str, raw: Optional[Dict[str, Any]] = None):
+ def __init__(self, client: FmdClient, fmd_id: str, raw: Optional[Dict[str, JSONType]] = None) -> None:
self.client = client
self.id = fmd_id
- self.raw: Dict[str, Any] = raw or {}
+ self.raw: Dict[str, JSONType] = raw or {}
self.name = self.raw.get("name")
self.cached_location: Optional[Location] = None
self._last_refresh = None
@@ -53,7 +53,7 @@ async def get_location(self, *, force: bool = False) -> Optional[Location]:
return self.cached_location
async def get_history(
- self, start: Optional[Any] = None, end: Optional[Any] = None, limit: int = -1
+ self, start: Optional[Union[int, datetime]] = None, end: Optional[Union[int, datetime]] = None, limit: int = -1
) -> AsyncIterator[Location]:
"""
Iterate historical locations. Uses client.get_locations() under the hood.
@@ -77,44 +77,6 @@ async def get_history(
async def play_sound(self) -> bool:
return await self.client.send_command("ring")
- async def take_front_photo(self) -> bool:
- warnings.warn(
- "Device.take_front_photo() is deprecated; use take_front_picture()",
- DeprecationWarning,
- stacklevel=2,
- )
- return await self.take_front_picture()
-
- async def take_rear_photo(self) -> bool:
- warnings.warn(
- "Device.take_rear_photo() is deprecated; use take_rear_picture()",
- DeprecationWarning,
- stacklevel=2,
- )
- return await self.take_rear_picture()
-
- async def fetch_pictures(self, num_to_get: int = -1) -> List[Dict[str, Any]]:
- warnings.warn(
- "Device.fetch_pictures() is deprecated; use get_picture_blobs()",
- DeprecationWarning,
- stacklevel=2,
- )
- return await self.get_picture_blobs(num_to_get=num_to_get)
-
- async def download_photo(self, picture_blob_b64: str) -> PhotoResult:
- """
- Decrypt a picture blob and return binary PhotoResult.
-
- The fmd README says picture data is double-encoded: encrypted blob -> base64 string -> image bytes.
- We decrypt the blob to get a base64-encoded image string; decode that to bytes and return.
- """
- warnings.warn(
- "Device.download_photo() is deprecated; use decode_picture()",
- DeprecationWarning,
- stacklevel=2,
- )
- return await self.decode_picture(picture_blob_b64)
-
async def take_front_picture(self) -> bool:
"""Request a picture from the front camera."""
return await self.client.take_picture("front")
@@ -123,27 +85,26 @@ async def take_rear_picture(self) -> bool:
"""Request a picture from the rear camera."""
return await self.client.take_picture("back")
- async def get_pictures(self, num_to_get: int = -1) -> List[Dict[str, Any]]:
- """Deprecated: use get_picture_blobs()."""
- warnings.warn(
- "Device.get_pictures() is deprecated; use get_picture_blobs()",
- DeprecationWarning,
- stacklevel=2,
- )
- return await self.get_picture_blobs(num_to_get=num_to_get)
-
- async def get_picture(self, picture_blob_b64: str) -> PhotoResult:
- """Deprecated: use decode_picture()."""
- warnings.warn(
- "Device.get_picture() is deprecated; use decode_picture()",
- DeprecationWarning,
- stacklevel=2,
- )
- return await self.decode_picture(picture_blob_b64)
-
- async def get_picture_blobs(self, num_to_get: int = -1) -> List[Dict[str, Any]]:
- """Get raw picture blobs (base64-encoded encrypted strings) from the server."""
- return await self.client.get_pictures(num_to_get=num_to_get)
+ async def get_picture_blobs(self, num_to_get: int = -1) -> List[JSONType]:
+ """Get raw picture blobs (usually a list of dicts returned by the server).
+
+ The client may return lists of mixed JSON values; coerce/filter here so callers
+ (and the annotated return type) are guaranteed a list of dicts.
+ """
+ # Return raw server values (strings, dicts, etc.) — callers like decode_picture
+ # expect string blobs and some server versions return dicts; preserve this.
+ raw = await self.client.get_pictures(num_to_get=num_to_get)
+ return list(raw)
+
+ async def get_picture_metadata(self, num_to_get: int = -1) -> List[PictureMetadata]:
+ """Return only picture entries that are JSON objects (dicts) — strongly typed metadata.
+
+ This method intentionally filters the raw values returned by `get_picture_blobs` and
+ yields only mappings (dicts), which callers can rely on for metadata fields.
+ """
+ raw = await self.client.get_pictures(num_to_get=num_to_get)
+ metadata: List[PictureMetadata] = [cast(PictureMetadata, p) for p in raw if isinstance(p, dict)]
+ return metadata
async def decode_picture(self, picture_blob_b64: str) -> PhotoResult:
"""Decrypt and decode a single picture blob into a PhotoResult."""
diff --git a/fmd_api/models.py b/fmd_api/models.py
index 78b72af..ccaedd3 100644
--- a/fmd_api/models.py
+++ b/fmd_api/models.py
@@ -1,6 +1,7 @@
from dataclasses import dataclass
from datetime import datetime, timezone
-from typing import Optional, Dict, Any, Union
+from typing import Optional, Dict, Union
+from .types import JSONType
import json as _json
@@ -15,10 +16,10 @@ class Location:
heading_deg: Optional[float] = None
battery_pct: Optional[int] = None
provider: Optional[str] = None
- raw: Optional[Dict[str, Any]] = None
+ raw: Optional[Dict[str, JSONType]] = None
@classmethod
- def from_json(cls, json: Union[str, Dict[str, Any]]) -> "Location":
+ def from_json(cls, json: Union[str, Dict[str, JSONType]]) -> "Location":
"""Construct a Location from a JSON dict or JSON string.
Expected fields (from server payloads):
@@ -68,4 +69,4 @@ class PhotoResult:
data: bytes
mime_type: str
timestamp: datetime
- raw: Optional[Dict[str, Any]] = None
+ raw: Optional[Dict[str, JSONType]] = None
diff --git a/fmd_api/types.py b/fmd_api/types.py
new file mode 100644
index 0000000..fa80ed3
--- /dev/null
+++ b/fmd_api/types.py
@@ -0,0 +1,36 @@
+"""Shared typing helpers used across the fmd_api package.
+
+This module centralizes JSON-like typings and commonly used typed dictionaries so
+other modules in the package can import concrete types rather than using
+unstructured Any in many places.
+"""
+from __future__ import annotations
+
+from typing import Dict, List, Union, TypedDict, Optional
+
+
+# Recursive JSON-ish type used for payloads / returned JSON values
+JSONType = Union[Dict[str, "JSONType"], List["JSONType"], str, int, float, bool, None]
+
+
+class AuthArtifacts(TypedDict, total=False):
+ base_url: str
+ fmd_id: Optional[str]
+ access_token: Optional[str]
+ private_key: str
+ password_hash: Optional[str]
+ session_duration: int
+ token_issued_at: Optional[float]
+
+
+class PictureMetadata(TypedDict, total=False):
+ """Common picture metadata fields returned by server endpoints.
+
+ The server returns varying shapes; this TypedDict captures common fields.
+ Fields are optional because older/newer server versions may differ.
+ """
+ id: int
+ date: int
+ filename: str
+ size: int
+ index: int
diff --git a/pyproject.toml b/pyproject.toml
index 0554401..29f174f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "fmd_api"
-version = "2.0.5"
+version = "2.0.6"
authors = [{name = "devinslick"}]
description = "A Python client for the FMD (Find My Device) server API"
readme = "README.md"
@@ -96,9 +96,4 @@ warn_no_return = true
warn_unreachable = true
strict_equality = true
show_error_codes = true
-# Allow Any in tests for now
-[[tool.mypy.overrides]]
-module = "tests.*"
-ignore_errors = true
-disallow_untyped_defs = false
-disallow_incomplete_defs = false
+# Do not special-case the tests directory (enforce strict typing across the whole repo)
diff --git a/tests/unit/test_coverage_improvements.py b/tests/unit/test_coverage_improvements.py
index 981caad..ae0dc1d 100644
--- a/tests/unit/test_coverage_improvements.py
+++ b/tests/unit/test_coverage_improvements.py
@@ -437,7 +437,7 @@ async def test_device_download_photo_decode_error():
blob_b64 = base64.b64encode(blob).decode("utf-8")
with pytest.raises(OperationError, match="Failed to decode picture blob"):
- await device.download_photo(blob_b64)
+ await device.decode_picture(blob_b64)
@pytest.mark.asyncio
@@ -491,6 +491,25 @@ async def test_retry_after_header_parsing_indirectly():
await client.close()
+@pytest.mark.asyncio
+async def test_rate_limit_exhausted_retries_direct():
+ """Directly exercise the 429 exhaustion path in _make_api_request by setting max_retries=0."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ # Return a 429 and no retry attempts allowed
+ m.get("https://fmd.example.com/api/v1/test", status=429)
+
+ try:
+ with pytest.raises(FmdApiException, match=r"Rate limited \(429\)"):
+ await client._make_api_request("GET", "/api/v1/test", {"IDT": "token", "Data": ""}, max_retries=0)
+ finally:
+ await client.close()
+
+
# ==========================================
# Additional edge cases
# ==========================================
@@ -522,30 +541,6 @@ async def test_send_command_with_missing_private_key():
await client.close()
-@pytest.mark.asyncio
-async def test_device_fetch_pictures_deprecated():
- """Test fetch_pictures() deprecated wrapper emits warning."""
- client = FmdClient("https://fmd.example.com")
- client.access_token = "token"
- device = Device(client, "test-device")
-
- with aioresponses() as m:
- m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": ["blob1", "blob2"]})
-
- try:
- import warnings
-
- with warnings.catch_warnings(record=True) as w:
- warnings.simplefilter("always")
- result = await device.fetch_pictures(2)
- assert len(w) == 1
- assert issubclass(w[0].category, DeprecationWarning)
- assert "fetch_pictures() is deprecated" in str(w[0].message)
- assert len(result) == 2
- finally:
- await client.close()
-
-
@pytest.mark.asyncio
async def test_client_error_generic():
"""Test generic ClientError handling."""
@@ -1115,3 +1110,282 @@ async def test_non_json_response_with_expect_json_false():
assert result == "plain text response"
finally:
await client.close()
+
+
+@pytest.mark.asyncio
+async def test_json_list_fallback_to_text_empty_body():
+ """When server returns an empty body with application/json, parsing should fall back and return empty text."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ # Empty body with JSON content type will cause JSONDecodeError and empty text -> triggers warning branch
+ m.put("https://fmd.example.com/api/v1/salt", body="", content_type="application/json")
+
+ try:
+ result = await client._make_api_request("PUT", "/api/v1/salt", {"IDT": "test", "Data": ""})
+ assert result == ""
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_make_api_request_log_keys_when_debug_enabled():
+ """Ensure the JSON dict-keys logging branch executes when DEBUG enabled (coverage lines 512-513)."""
+ import logging
+
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+ await client._ensure_session()
+
+ # Enable debug for the module logger to exercise the keys logging
+ logger = logging.getLogger("fmd_api.client")
+ old_level = logger.level
+ logger.setLevel(logging.DEBUG)
+
+ with aioresponses() as m:
+ m.put(
+ "https://fmd.example.com/api/v1/salt",
+ payload={"Data": {"x": 1, "y": 2}},
+ content_type="application/json",
+ )
+
+ try:
+ result = await client._make_api_request("PUT", "/api/v1/salt", {"IDT": "test", "Data": ""})
+ assert result == {"x": 1, "y": 2}
+ finally:
+ logger.setLevel(old_level)
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_get_locations_unexpected_size_type_raises():
+ """If locationDataSize returns a non-str/non-int, get_locations should raise FmdApiException (line 569)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ # Return Data as an object which isn't valid for size
+ m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": {"bad": True}})
+
+ try:
+ with pytest.raises(FmdApiException, match="Unexpected value for locationDataSize"):
+ await client.get_locations()
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_get_locations_skip_non_string_blob():
+ """get_locations should skip non-string blobs and warn (line 587)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "2"})
+ # Return a dict for index 0 and valid string for index 1
+ m.put("https://fmd.example.com/api/v1/location", payload={"Data": {"meta": 1}})
+ m.put("https://fmd.example.com/api/v1/location", payload={"Data": "validblob"})
+
+ try:
+ res = await client.get_locations(num_to_get=-1)
+ # Should only include the string blob
+ assert res == ["validblob"]
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_get_pictures_unexpected_type_returns_empty():
+ """get_pictures should return [] for unexpected non-list responses (lines 654-655)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ # Return a dict (not a list) in Data
+ m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": {"meta": "bad"}})
+
+ try:
+ res = await client.get_pictures()
+ assert res == []
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_export_zip_non_text_location_blob(monkeypatch):
+ """export_data_zip should record an error for non-text location blob (lines 718-720)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+
+ # Set up private key for decryption
+ private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend())
+ client.private_key = private_key
+
+ await client._ensure_session()
+
+ # Patch get_locations to return a non-text blob (async function)
+ async def fake_get_locations(*args, **kwargs):
+ return [{"meta": 1}]
+
+ async def fake_get_pictures(*args, **kwargs):
+ return []
+
+ monkeypatch.setattr(client, "get_locations", fake_get_locations)
+ monkeypatch.setattr(client, "get_pictures", fake_get_pictures)
+
+ try:
+ import tempfile
+
+ with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
+ output_path = tmp.name
+
+ await client.export_data_zip(output_path, include_pictures=False)
+
+ # locations.json should contain an error entry for index 0
+ import json
+ import zipfile
+
+ with zipfile.ZipFile(output_path, "r") as zf:
+ locations = json.loads(zf.read("locations.json"))
+ assert locations[0]["error"] == "invalid blob type"
+
+ import os
+ if os.path.exists(output_path):
+ os.unlink(output_path)
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_safe_read_text_returns_none_on_exception():
+ """_safe_read_text returns None if resp.text() raises an exception (lines 899-900)."""
+ from fmd_api.client import _safe_read_text
+
+ class DummyResp:
+ async def text(self):
+ raise RuntimeError("boom")
+
+ resp = DummyResp()
+ result = await _safe_read_text(resp)
+ assert result is None
+
+
+@pytest.mark.asyncio
+async def test_make_api_request_keyerror_wrapped():
+ """If session.request raises a KeyError, it should be wrapped as FmdApiException (lines 552-553)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ m.get("https://fmd.example.com/api/v1/test", exception=KeyError("boom"))
+
+ try:
+ with pytest.raises(FmdApiException, match="Failed to parse server response"):
+ await client._make_api_request("GET", "/api/v1/test", {"IDT": "token", "Data": ""})
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_export_zip_non_text_picture_blob(monkeypatch):
+ """export_data_zip should record an error for non-text picture blobs (lines 735-737)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+
+ # Set up private key (not used because get_pictures is patched)
+ private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend())
+ client.private_key = private_key
+
+ # Patch get_locations to be empty list and get_pictures to return non-text blob
+ async def fake_get_locations(*args, **kwargs):
+ return []
+
+ async def fake_get_pictures(*args, **kwargs):
+ return [{"meta": "not-a-string"}]
+
+ monkeypatch.setattr(client, "get_locations", fake_get_locations)
+ monkeypatch.setattr(client, "get_pictures", fake_get_pictures)
+
+ try:
+ import tempfile
+
+ with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
+ output_path = tmp.name
+
+ await client.export_data_zip(output_path, include_pictures=True)
+
+ import json
+ import zipfile
+ import os
+
+ with zipfile.ZipFile(output_path, "r") as zf:
+ # manifest should exist and indicate error for index 0
+ manifest = json.loads(zf.read("pictures/manifest.json"))
+ assert manifest[0]["error"] == "invalid blob type"
+
+ if os.path.exists(output_path):
+ os.unlink(output_path)
+ finally:
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_export_zip_jpeg_detection():
+ """Test JPEG magic byte detection in export_data_zip to exercise the JPEG branch (line ~749)."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+
+ # Set up private key
+ private_key = rsa.generate_private_key(public_exponent=65537, key_size=3072, backend=default_backend())
+ client.private_key = private_key
+
+ # Create JPEG-like image bytes
+ jpeg_bytes = b"\xff\xd8\xff" + b"\x00" * 64
+ inner_b64 = base64.b64encode(jpeg_bytes).decode("utf-8")
+
+ session_key = b"\x01" * 32
+ aesgcm = AESGCM(session_key)
+ iv = b"\x02" * 12
+ ciphertext = aesgcm.encrypt(iv, inner_b64.encode("utf-8"), None)
+
+ public_key = private_key.public_key()
+ session_key_packet = public_key.encrypt(
+ session_key,
+ asym_padding.OAEP(mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None),
+ )
+
+ blob = session_key_packet + iv + ciphertext
+ blob_b64 = base64.b64encode(blob).decode("utf-8")
+
+ await client._ensure_session()
+
+ with aioresponses() as m:
+ m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "0"})
+ m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": [blob_b64]})
+
+ try:
+ import tempfile
+ import zipfile
+ import os
+
+ with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
+ output_path = tmp.name
+
+ result = await client.export_data_zip(output_path, include_pictures=True)
+ assert result == output_path
+
+ # Verify ZIP contains a file with .jpg extension
+ with zipfile.ZipFile(output_path, "r") as zf:
+ files = zf.namelist()
+ assert any(f.endswith(".jpg") for f in files)
+
+ if os.path.exists(output_path):
+ os.unlink(output_path)
+ finally:
+ await client.close()
diff --git a/tests/unit/test_deprecated.py b/tests/unit/test_deprecated.py
deleted file mode 100644
index 0daa67c..0000000
--- a/tests/unit/test_deprecated.py
+++ /dev/null
@@ -1,86 +0,0 @@
-import pytest
-from unittest.mock import AsyncMock, MagicMock
-from fmd_api.device import Device
-from fmd_api.client import FmdClient
-
-
-@pytest.mark.asyncio
-async def test_deprecated_take_front_photo():
- client = MagicMock(spec=FmdClient)
- device = Device(client, "test-device")
- device.take_front_picture = AsyncMock(return_value=True)
-
- with pytest.warns(DeprecationWarning, match="take_front_photo.*deprecated"):
- result = await device.take_front_photo()
-
- assert result is True
- device.take_front_picture.assert_awaited_once()
-
-
-@pytest.mark.asyncio
-async def test_deprecated_take_rear_photo():
- client = MagicMock(spec=FmdClient)
- device = Device(client, "test-device")
- device.take_rear_picture = AsyncMock(return_value=True)
-
- with pytest.warns(DeprecationWarning, match="take_rear_photo.*deprecated"):
- result = await device.take_rear_photo()
-
- assert result is True
- device.take_rear_picture.assert_awaited_once()
-
-
-@pytest.mark.asyncio
-async def test_deprecated_fetch_pictures():
- client = MagicMock(spec=FmdClient)
- device = Device(client, "test-device")
- expected = [{"id": "123"}]
- device.get_picture_blobs = AsyncMock(return_value=expected)
-
- with pytest.warns(DeprecationWarning, match="fetch_pictures.*deprecated"):
- result = await device.fetch_pictures(num_to_get=5)
-
- assert result == expected
- device.get_picture_blobs.assert_awaited_once_with(num_to_get=5)
-
-
-@pytest.mark.asyncio
-async def test_deprecated_get_pictures():
- client = MagicMock(spec=FmdClient)
- device = Device(client, "test-device")
- expected = [{"id": "123"}]
- device.get_picture_blobs = AsyncMock(return_value=expected)
-
- with pytest.warns(DeprecationWarning, match="get_pictures.*deprecated"):
- result = await device.get_pictures(num_to_get=5)
-
- assert result == expected
- device.get_picture_blobs.assert_awaited_once_with(num_to_get=5)
-
-
-@pytest.mark.asyncio
-async def test_deprecated_download_photo():
- client = MagicMock(spec=FmdClient)
- device = Device(client, "test-device")
- expected = MagicMock() # PhotoResult
- device.decode_picture = AsyncMock(return_value=expected)
-
- with pytest.warns(DeprecationWarning, match="download_photo.*deprecated"):
- result = await device.download_photo("blob")
-
- assert result == expected
- device.decode_picture.assert_awaited_once_with("blob")
-
-
-@pytest.mark.asyncio
-async def test_deprecated_get_picture():
- client = MagicMock(spec=FmdClient)
- device = Device(client, "test-device")
- expected = MagicMock() # PhotoResult
- device.decode_picture = AsyncMock(return_value=expected)
-
- with pytest.warns(DeprecationWarning, match="get_picture.*deprecated"):
- result = await device.get_picture("blob")
-
- assert result == expected
- device.decode_picture.assert_awaited_once_with("blob")
diff --git a/tests/unit/test_device.py b/tests/unit/test_device.py
index c90c103..ee4d511 100644
--- a/tests/unit/test_device.py
+++ b/tests/unit/test_device.py
@@ -85,6 +85,36 @@ def decrypt(self, packet, padding_obj):
await client.close()
+@pytest.mark.asyncio
+async def test_get_picture_blobs_and_metadata_mix(monkeypatch):
+ """Ensure get_picture_blobs returns raw values (strings/dicts) and get_picture_metadata filters to dicts."""
+ client = FmdClient("https://fmd.example.com")
+ client.access_token = "token"
+
+ # Blob string and metadata dict
+ blob_str = "somestringblob"
+ metadata = {"id": 42, "date": 1600000000000}
+
+ await client._ensure_session()
+ device = Device(client, "alice")
+
+ with aioresponses() as m:
+ # Register the same mock for two calls (get_picture_blobs then get_picture_metadata)
+ m.put("https://fmd.example.com/api/v1/pictures", payload=[blob_str, metadata], repeat=True)
+ try:
+ raw = await device.get_picture_blobs()
+ assert len(raw) == 2
+ assert raw[0] == blob_str
+ assert isinstance(raw[1], dict)
+
+ # metadata method should only return dicts
+ md = await device.get_picture_metadata()
+ assert len(md) == 1
+ assert md[0]["id"] == 42
+ finally:
+ await client.close()
+
+
@pytest.mark.asyncio
async def test_device_command_wrappers():
"""Test Device command wrapper methods."""
@@ -107,8 +137,8 @@ def sign(self, message_bytes, pad, algo):
m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
try:
assert await device.play_sound() is True
- assert await device.take_front_photo() is True
- assert await device.take_rear_photo() is True
+ assert await device.take_front_picture() is True
+ assert await device.take_rear_picture() is True
assert await device.lock() is True
finally:
await client.close()
diff --git a/tests/unit/test_resume.py b/tests/unit/test_resume.py
index c6f5ad6..8fc0870 100644
--- a/tests/unit/test_resume.py
+++ b/tests/unit/test_resume.py
@@ -126,6 +126,50 @@ async def test_from_auth_artifacts_missing_fields():
await FmdClient.from_auth_artifacts(incomplete)
+@pytest.mark.asyncio
+async def test_from_auth_artifacts_invalid_types():
+ """If required artifact keys are present but of incorrect types, raise ValueError."""
+ # base_url is an int -> invalid
+ artifacts = {
+ "base_url": 123,
+ "fmd_id": "alice",
+ "access_token": "token",
+ "private_key": "pem-string",
+ }
+
+ with pytest.raises(ValueError, match="Missing or invalid artifact fields"):
+ await FmdClient.from_auth_artifacts(artifacts)
+
+
+@pytest.mark.asyncio
+async def test_from_auth_artifacts_nonint_session_duration_defaults():
+ """If session_duration isn't an int it should fall back to the default (3600)."""
+ from cryptography.hazmat.primitives.asymmetric import rsa
+ from cryptography.hazmat.primitives import serialization
+
+ # Generate a valid key PEM so resume() can accept the artifact
+ key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+ pem = key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.PKCS8,
+ encryption_algorithm=serialization.NoEncryption(),
+ ).decode("utf-8")
+
+ artifacts = {
+ "base_url": "https://fmd.example.com",
+ "fmd_id": "alice",
+ "access_token": "tkn1",
+ "private_key": pem,
+ "session_duration": "not-an-int",
+ }
+
+ resumed = await FmdClient.from_auth_artifacts(artifacts)
+ try:
+ assert resumed.session_duration == 3600
+ finally:
+ await resumed.close()
+
+
@pytest.mark.asyncio
async def test_export_artifacts_without_private_key():
"""Test export_auth_artifacts raises when private key not loaded."""