Add code test coverage

Author: devinslick

fmd_api/client.py

@@ -282,11 +282,18 @@ async def get_pictures(self, num_to_get: int = -1) -> List[Any]:
             await self._ensure_session()
             async with self._session.put(f"{self.base_url}/api/v1/pictures", json={"IDT": self.access_token, "Data": ""}) as resp:
                 resp.raise_for_status()
-                all_pictures = await resp.json()
+                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
         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 []
+
         if num_to_get == -1:
             log.info(f"Found {len(all_pictures)} pictures to download.")
             return all_pictures

tests/unit/test_client.py

@@ -301,3 +301,174 @@ async def test_empty_location_response():
             assert locs == []
         finally:
             await client.close()
+
+@pytest.mark.asyncio
+async def test_get_all_locations():
+    """Test get_locations with num_to_get=-1 fetches all locations."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+    
+    # Create 3 location blobs
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    
+    blobs = []
+    for i in range(3):
+        iv = bytes([i+1] * 12)
+        plaintext = json.dumps({"lat": float(i), "lon": float(i*10), "date": 1600000000000, "bat": 80}).encode('utf-8')
+        ciphertext = aesgcm.encrypt(iv, plaintext, None)
+        blob = b'\xAA' * 384 + iv + ciphertext
+        blobs.append(base64.b64encode(blob).decode('utf-8').rstrip('='))
+    
+    await client._ensure_session()
+    
+    with aioresponses() as m:
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "3"})
+        # When fetching all, indices are 0, 1, 2
+        for i, blob_b64 in enumerate(blobs):
+            m.put("https://fmd.example.com/api/v1/location", payload={"Data": blob_b64})
+        
+        try:
+            locs = await client.get_locations(num_to_get=-1)
+            assert len(locs) == 3
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_skip_empty_locations():
+    """Test that skip_empty skips over empty blobs to find valid ones."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+    
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    iv = b'\x05' * 12
+    plaintext = b'{"lat":5.0,"lon":10.0,"date":1600000000000,"bat":90}'
+    ciphertext = aesgcm.encrypt(iv, plaintext, None)
+    blob = b'\xAA' * 384 + iv + ciphertext
+    blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
+    
+    await client._ensure_session()
+    
+    with aioresponses() as m:
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "3"})
+        # Index 2 (most recent): empty
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": ""})
+        # Index 1: empty
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": ""})
+        # Index 0: valid
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": blob_b64})
+        
+        try:
+            locs = await client.get_locations(num_to_get=1, skip_empty=True)
+            assert len(locs) == 1
+            assert locs[0] == blob_b64
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_picture_endpoint_error():
+    """Test error handling when pictures endpoint fails."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    await client._ensure_session()
+    
+    with aioresponses() as m:
+        # pictures endpoint returns 500 error
+        m.put("https://fmd.example.com/api/v1/pictures", status=500)
+        
+        try:
+            # Should return empty list on error (client logs warning)
+            pictures = await client.get_pictures()
+            assert pictures == []
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_multiple_commands_sequence():
+    """Test sending multiple commands in sequence."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    class DummySigner:
+        def sign(self, message_bytes, pad, algo):
+            return b"\xAB" * 64
+    client.private_key = DummySigner()
+    
+    await client._ensure_session()
+    
+    with aioresponses() as m:
+        # Mock multiple command requests
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        
+        try:
+            result1 = await client.send_command("ring")
+            assert result1 is True
+            
+            result2 = await client.send_command("lock")
+            assert result2 is True
+            
+            result3 = await client.send_command("locate")
+            assert result3 is True
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_get_pictures_pagination():
+    """Test get_pictures with num_to_get limit."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    await client._ensure_session()
+    
+    with aioresponses() as m:
+        # Mock pictures response with multiple pictures (already in order)
+        mock_pictures = [
+            {"id": 2, "date": 1600000002000},
+            {"id": 1, "date": 1600000001000},
+            {"id": 0, "date": 1600000000000}
+        ]
+        m.put("https://fmd.example.com/api/v1/pictures", payload={"Data": mock_pictures})
+        
+        try:
+            pictures = await client.get_pictures(num_to_get=2)
+            assert len(pictures) == 2
+            # Should get the 2 most recent (last 2, reversed)
+            # [-2:][::-1] on [2,1,0] gives [1,0] then reverses to [0,1]
+            assert pictures[0]["id"] == 0
+            assert pictures[1]["id"] == 1
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_authenticate_error_handling():
+    """Test authenticate with invalid credentials."""
+    client = FmdClient("https://fmd.example.com")
+    
+    await client._ensure_session()
+    
+    with aioresponses() as m:
+        # Mock authentication endpoints - first call is to /api/v1/salt
+        m.put("https://fmd.example.com/api/v1/salt", status=401)
+        
+        try:
+            from fmd_api.exceptions import FmdApiException
+            with pytest.raises(FmdApiException, match="API request failed for /api/v1/salt"):
+                await client.authenticate("bad_id", "bad_password", session_duration=3600)
+        finally:
+            await client.close()

tests/unit/test_device.py

@@ -235,3 +235,181 @@ def decrypt(self, packet, padding_obj):
             assert abs(loc3.lat - 15.0) < 1e-6
         finally:
             await client.close()
+
+@pytest.mark.asyncio
+async def test_device_cached_location_property():
+    """Test Device.cached_location property access."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+    
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    iv = b'\x06' * 12
+    plaintext = b'{"lat":20.0,"lon":30.0,"date":1600000000000,"bat":75}'
+    ciphertext = aesgcm.encrypt(iv, plaintext, None)
+    blob = b'\xAA' * 384 + iv + ciphertext
+    blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
+    
+    client.access_token = "token"
+    device = Device(client, "test-device")
+    
+    # Initially None
+    assert device.cached_location is None
+    
+    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={"Data": blob_b64})
+        try:
+            await device.refresh()
+            # Now should have cached location
+            assert device.cached_location is not None
+            assert abs(device.cached_location.lat - 20.0) < 1e-6
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_device_refresh_without_force():
+    """Test Device.refresh with force=False doesn't re-fetch if cached."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+    
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    iv = b'\x07' * 12
+    plaintext = b'{"lat":25.0,"lon":35.0,"date":1600000000000,"bat":85}'
+    ciphertext = aesgcm.encrypt(iv, plaintext, None)
+    blob = b'\xAA' * 384 + iv + ciphertext
+    blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
+    
+    client.access_token = "token"
+    device = Device(client, "test-device")
+    
+    with aioresponses() as m:
+        # Only one set of mocks
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "1"})
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": blob_b64})
+        try:
+            await device.refresh()
+            loc1 = device.cached_location
+            
+            # Second refresh without force should not make HTTP calls (mocks would fail if it did)
+            await device.refresh(force=False)
+            loc2 = device.cached_location
+            
+            assert loc1 is loc2  # Same cached object
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_device_picture_commands():
+    """Test Device picture-related command shortcuts."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    class DummySigner:
+        def sign(self, message_bytes, pad, algo):
+            return b"\xAB" * 64
+    client.private_key = DummySigner()
+    
+    await client._ensure_session()
+    device = Device(client, "test-device")
+    
+    with aioresponses() as m:
+        # take_front_photo
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        # take_rear_photo
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        
+        try:
+            result1 = await device.take_front_photo()
+            assert result1 is True
+            
+            result2 = await device.take_rear_photo()
+            assert result2 is True
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_device_lock_with_message():
+    """Test Device.lock with custom message."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    class DummySigner:
+        def sign(self, message_bytes, pad, algo):
+            return b"\xAB" * 64
+    client.private_key = DummySigner()
+    
+    await client._ensure_session()
+    device = Device(client, "test-device")
+    
+    with aioresponses() as m:
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        
+        try:
+            result = await device.lock("Please return this device")
+            assert result is True
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_device_multiple_history_calls():
+    """Test Device.get_history can be called multiple times."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b'\x00' * 32
+    client.private_key = DummyKey()
+    
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    session_key = b'\x00' * 32
+    aesgcm = AESGCM(session_key)
+    
+    blobs = []
+    for i in range(2):
+        iv = bytes([i+10] * 12)
+        plaintext = json.dumps({"lat": float(30+i), "lon": float(40+i), "date": 1600000000000 + i*1000, "bat": 80}).encode('utf-8')
+        ciphertext = aesgcm.encrypt(iv, plaintext, None)
+        blob = b'\xAA' * 384 + iv + ciphertext
+        blobs.append(base64.b64encode(blob).decode('utf-8').rstrip('='))
+    
+    client.access_token = "token"
+    device = Device(client, "test-device")
+    
+    with aioresponses() as m:
+        # First call to get_history
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "2"})
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": blobs[0]})
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": blobs[1]})
+        
+        # Second call to get_history
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "2"})
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": blobs[0]})
+        m.put("https://fmd.example.com/api/v1/location", payload={"Data": blobs[1]})
+        
+        try:
+            # First iteration
+            locs1 = []
+            async for loc in device.get_history(limit=2):
+                locs1.append(loc)
+            assert len(locs1) == 2
+            
+            # Second iteration (should work independently)
+            locs2 = []
+            async for loc in device.get_history(limit=2):
+                locs2.append(loc)
+            assert len(locs2) == 2
+            
+            # Both should have same data (different Location objects)
+            assert abs(locs1[0].lat - locs2[0].lat) < 1e-6
+        finally:
+            await client.close()