Cleanup and version update

Author: devinslick

examples/tests/test_scripts/unit_test_client.py

@@ -0,0 +1,87 @@
+import asyncio
+import json
+import base64
+from datetime import datetime, timezone
+
+import pytest
+import aiohttp
+from aioresponses import aioresponses
+
+from fmd_api.client import FmdClient
+from fmd_api.helpers import _pad_base64
+
+# NOTE: These tests validate behavior parity for the core HTTP flows using mocks.
+# They do not perform full Argon2/RSA cryptography verification, but they assert
+# that the client calls the expected endpoints and behaves like the original client.
+
+@pytest.mark.asyncio
+async def test_get_locations_and_decrypt(monkeypatch):
+    # Create a fake client and stub methods that require heavy crypto with small helpers.
+    client = FmdClient("https://fmd.example.com")
+    # Provide a dummy private_key with a decrypt method for testing
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            # Return a 32-byte AES session key for AESGCM, for tests we use 32 zero bytes
+            return b"\x00" * 32
+    client.private_key = DummyKey()
+
+    # Build a fake AES-GCM encrypted payload: we'll create plaintext b'{"lat":1.0,"lon":2.0,"date":1234,"bat":50}'
+    plaintext = b'{"lat":1.0,"lon":2.0,"date":1600000000000,"bat":50}'
+    # For the test, simulate AESGCM by encrypting with a known key using AESGCM class
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b"\x00" * 32
+    aesgcm = AESGCM(session_key)
+    iv = b"\x01" * 12
+    ciphertext = aesgcm.encrypt(iv, plaintext, None)
+    # Build blob: session_key_packet (RSA_KEY_SIZE_BYTES) + iv + ciphertext
+    session_key_packet = b"\xAA" * 384  # dummy RSA packet; DummyKey.decrypt ignores it
+    blob = session_key_packet + iv + ciphertext
+    blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
+
+    # Mock the endpoints used by get_locations:
+    with aioresponses() as m:
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "1"})
+        m.put("https://fmd.example.com/api/v1/location", payload=blob_b64)
+        client.access_token = "dummy-token"
+        locations = await client.get_locations(num_to_get=1)
+        assert len(locations) == 1
+        decrypted = client.decrypt_data_blob(locations[0])
+        assert b'"lat":1.0' in decrypted
+        assert b'"lon":2.0' in decrypted
+
+@pytest.mark.asyncio
+async def test_send_command_reauth(monkeypatch):
+    client = FmdClient("https://fmd.example.com")
+    # create a dummy private key with sign()
+    class DummySigner:
+        def sign(self, message_bytes, pad, algo):
+            return b"\xAB" * 64
+    client.private_key = DummySigner()
+    client._fmd_id = "id"
+    client._password = "pw"
+    client.access_token = "old-token"
+
+    with aioresponses() as m:
+        # First POST returns 401 -> client should re-authenticate
+        m.post("https://fmd.example.com/api/v1/command", status=401)
+        # When authenticate is called during reauth, stub the internal calls:
+        async def fake_authenticate(fmd_id, password, session_duration):
+            client.access_token = "new-token"
+        monkeypatch.setattr(client, "authenticate", fake_authenticate)
+        # Second attempt should now succeed
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        res = await client.send_command("ring")
+        assert res is True
+
+@pytest.mark.asyncio
+async def test_export_data_zip_stream(monkeypatch, tmp_path):
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    small_zip = b'PK\x03\x04' + b'\x00' * 100
+    with aioresponses() as m:
+        m.post("https://fmd.example.com/api/v1/exportData", body=small_zip, status=200)
+        out_file = tmp_path / "export.zip"
+        await client.export_data_zip(str(out_file))
+        assert out_file.exists()
+        content = out_file.read_bytes()
+        assert content.startswith(b'PK\x03\x04')

examples/tests/test_scripts/unit_test_device.py

@@ -0,0 +1,73 @@
+import asyncio
+import base64
+import json
+from datetime import datetime, timezone
+
+import pytest
+from aioresponses import aioresponses
+
+from fmd_api.client import FmdClient
+from fmd_api.device import Device
+
+@pytest.mark.asyncio
+async def test_device_refresh_and_get_location(monkeypatch):
+    client = FmdClient("https://fmd.example.com")
+    # Dummy private_key decrypt path (reuse approach from client tests)
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+
+    # Create a simple AES-GCM encrypted location blob (same scheme as client test)
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    iv = b'\x02' * 12
+    plaintext = b'{"lat":10.0,"lon":20.0,"date":1600000000000,"bat":80}'
+    ciphertext = aesgcm.encrypt(iv, plaintext, None)
+    blob = b'\xAA' * 384 + iv + ciphertext
+    blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
+
+    with aioresponses() as m:
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "1"})
+        m.put("https://fmd.example.com/api/v1/location", payload=blob_b64)
+        client.access_token = "token"
+        device = Device(client, "alice")
+        await device.refresh()
+        loc = await device.get_location()
+        assert loc is not None
+        assert abs(loc.lat - 10.0) < 1e-6
+        assert abs(loc.lon - 20.0) < 1e-6
+
+@pytest.mark.asyncio
+async def test_device_fetch_and_download_picture(monkeypatch):
+    client = FmdClient("https://fmd.example.com")
+    # Provide dummy private key that decrypts session packet into all-zero key
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+
+    # Prepare an "encrypted blob" that after decrypt yields a base64 image string.
+    # For test simplicity, we'll make decrypted payload the base64 of b'PNGDATA'
+    inner_image = base64.b64encode(b'PNGDATA').decode('utf-8')
+    # Encrypt inner_image using AESGCM with zero key
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    iv = b'\x03' * 12
+    ciphertext = aesgcm.encrypt(iv, inner_image.encode('utf-8'), None)
+    blob = b'\xAA' * 384 + iv + ciphertext
+    blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
+
+    with aioresponses() as m:
+        # get_pictures endpoint returns a JSON list; emulate simple list containing our blob
+        m.put("https://fmd.example.com/api/v1/pictures", payload=[blob_b64])
+        client.access_token = "token"
+        device = Device(client, "alice")
+        pics = await device.fetch_pictures()
+        assert len(pics) == 1
+        # download the picture and verify we got PNGDATA bytes
+        photo = await device.download_photo(pics[0])
+        assert photo.data == b'PNGDATA'
+        assert photo.mime_type.startswith("image/")

fmd_api/_version.py

@@ -1 +1 @@
-__version__ = "2.0.0-dev4"
+__version__ = "2.0.0-dev5"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fmd_api"
-version = "2.0.0.dev4"
+version = "2.0.0.dev5"
 authors = [{name = "devinslick"}]
 description = "A Python client for the FMD (Find My Device) server API"
 readme = "README.md"
@@ -32,12 +32,6 @@ dependencies = [
 requires = ["setuptools>=61.0", "wheel"]
 build-backend = "setuptools.build_meta"
 
-[tool.poetry.dev-dependencies]
-pytest = "^7.0"
-pytest-asyncio = "^0.20"
-aioresponses = "^0.9"
-
-
 [project.urls]
 Homepage = "https://github.com/devinslick/fmd_api"
 Repository = "https://github.com/devinslick/fmd_api"
@@ -48,6 +42,7 @@ Documentation = "https://github.com/devinslick/fmd_api#readme"
 dev = [
     "pytest>=7.0",
     "pytest-asyncio",
+    "aioresponses>=0.7.0",
     "black",
     "flake8",
     "mypy",