Fix bug prevented fetching locations when only 1 location exists; add test coverage

Author: devinslick

fmd_api/client.py

@@ -251,7 +251,7 @@ async def get_locations(self, num_to_get: int = -1, skip_empty: bool = True, max
             start_index = size - 1
 
             if skip_empty:
-                indices = range(start_index, max(0, start_index - max_attempts), -1)
+                indices = range(start_index, max(-1, start_index - max_attempts), -1)
                 log.info(f"Will search for {num_to_download} non-empty location(s) starting from index {start_index}")
             else:
                 end_index = size - num_to_download

tests/unit/test_client.py

@@ -39,10 +39,13 @@ def decrypt(self, packet, padding_obj):
     blob_b64 = base64.b64encode(blob).decode('utf-8').rstrip('=')
 
     # Mock the endpoints used by get_locations:
+    client.access_token = "dummy-token"
+    # Ensure session is created before entering aioresponses context
+    await client._ensure_session()
+    
     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})
-        client.access_token = "dummy-token"
         try:
             locations = await client.get_locations(num_to_get=1)
             assert len(locations) == 1
@@ -94,3 +97,207 @@ async def test_export_data_zip_stream(monkeypatch, tmp_path):
             assert content.startswith(b'PK\x03\x04')
         finally:
             await client.close()
+
+@pytest.mark.asyncio
+async def test_take_picture_validation():
+    """Test take_picture validates camera parameter."""
+    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()
+    
+    with aioresponses() as m:
+        # Valid cameras should work
+        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:
+            assert await client.take_picture("front") is True
+            assert await client.take_picture("back") is True
+        finally:
+            await client.close()
+    
+    # Invalid camera should raise ValueError
+    client2 = FmdClient("https://fmd.example.com")
+    client2.access_token = "token"
+    client2.private_key = DummySigner()
+    try:
+        with pytest.raises(ValueError, match="Invalid camera.*Must be 'front' or 'back'"):
+            await client2.take_picture("rear")
+    finally:
+        await client2.close()
+
+@pytest.mark.asyncio
+async def test_set_ringer_mode_validation():
+    """Test set_ringer_mode validates mode parameter."""
+    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()
+    
+    with aioresponses() as m:
+        # Valid modes should work
+        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:
+            assert await client.set_ringer_mode("normal") is True
+            assert await client.set_ringer_mode("vibrate") is True
+            assert await client.set_ringer_mode("silent") is True
+        finally:
+            await client.close()
+    
+    # Invalid mode should raise ValueError
+    client2 = FmdClient("https://fmd.example.com")
+    client2.access_token = "token"
+    client2.private_key = DummySigner()
+    try:
+        with pytest.raises(ValueError, match="Invalid ringer mode.*Must be"):
+            await client2.set_ringer_mode("loud")
+    finally:
+        await client2.close()
+
+@pytest.mark.asyncio
+async def test_request_location_providers():
+    """Test request_location with different providers."""
+    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()
+    
+    with aioresponses() as m:
+        # Mock all provider 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")
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        try:
+            assert await client.request_location("all") is True
+            assert await client.request_location("gps") is True
+            assert await client.request_location("cell") is True
+            assert await client.request_location("last") is True
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_set_bluetooth_and_dnd():
+    """Test set_bluetooth and set_do_not_disturb commands."""
+    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()
+    
+    with aioresponses() as m:
+        # Mock commands
+        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")
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        try:
+            assert await client.set_bluetooth(True) is True
+            assert await client.set_bluetooth(False) is True
+            assert await client.set_do_not_disturb(True) is True
+            assert await client.set_do_not_disturb(False) is True
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_get_device_stats():
+    """Test get_device_stats sends stats command."""
+    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()
+    
+    with aioresponses() as m:
+        m.post("https://fmd.example.com/api/v1/command", status=200, body="OK")
+        try:
+            assert await client.get_device_stats() is True
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_decrypt_data_blob_too_small():
+    """Test decrypt_data_blob raises FmdApiException for small blobs."""
+    from fmd_api.exceptions import FmdApiException
+    
+    client = FmdClient("https://fmd.example.com")
+    class DummyKey:
+        def decrypt(self, packet, padding_obj):
+            return b"\x00" * 32
+    client.private_key = DummyKey()
+    
+    # Blob must be at least RSA_KEY_SIZE_BYTES (384) + AES_GCM_IV_SIZE_BYTES (12) = 396 bytes
+    too_small = base64.b64encode(b"x" * 100).decode('utf-8')
+    
+    with pytest.raises(FmdApiException, match="Blob too small for decryption"):
+        client.decrypt_data_blob(too_small)
+
+@pytest.mark.asyncio
+async def test_get_pictures_direct():
+    """Test get_pictures endpoint directly."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    with aioresponses() as m:
+        # Mock pictures endpoint returning list of blobs
+        m.put("https://fmd.example.com/api/v1/pictures", payload=["blob1", "blob2", "blob3"])
+        try:
+            pics = await client.get_pictures(num_to_get=2)
+            assert len(pics) == 2
+            # Should get the 2 most recent (last 2 in reverse)
+            assert pics == ["blob3", "blob2"]
+        finally:
+            await client.close()
+
+@pytest.mark.asyncio
+async def test_http_error_handling():
+    """Test client handles various HTTP errors."""
+    from fmd_api.exceptions import FmdApiException
+    
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    # Test 404
+    with aioresponses() as m:
+        m.put("https://fmd.example.com/api/v1/locationDataSize", status=404)
+        try:
+            with pytest.raises(FmdApiException):
+                await client.get_locations()
+        finally:
+            await client.close()
+    
+    # Test 500
+    client2 = FmdClient("https://fmd.example.com")
+    client2.access_token = "token"
+    with aioresponses() as m:
+        m.put("https://fmd.example.com/api/v1/locationDataSize", status=500)
+        try:
+            with pytest.raises(FmdApiException):
+                await client2.get_locations()
+        finally:
+            await client2.close()
+
+@pytest.mark.asyncio
+async def test_empty_location_response():
+    """Test handling of empty location data."""
+    client = FmdClient("https://fmd.example.com")
+    client.access_token = "token"
+    
+    with aioresponses() as m:
+        # Server reports 0 locations
+        m.put("https://fmd.example.com/api/v1/locationDataSize", payload={"Data": "0"})
+        try:
+            locs = await client.get_locations()
+            assert locs == []
+        finally:
+            await client.close()