Skip to content

Commit 8c861b4

Browse files
devinslickdevinslick
committed
Add test scripts, attempt to fix export
1 parent 7569874 commit 8c861b4

File tree

11 files changed

+340
-11
lines changed

11 files changed

+340
-11
lines changed

‎.gitignore‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,4 +40,7 @@ env/
4040

4141
# FMD Server and android app files
4242
fmd-server/
43-
fmd-android/
43+
fmd-android/
44+
45+
#credentials file
46+
examples/tests/credentials.txt

‎examples/tests/credentials.txt.example‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
# Credentials file for fmd_api test scripts.
2+
# Save this file as 'credentials.txt' (same directory as the test scripts) and DO NOT commit real credentials.
3+
#
4+
# Format: one key=value per line. Required keys:
5+
# BASE_URL - base URL of the FMD server (e.g. https://fmd.example.com)
6+
# FMD_ID - your fmd id (account/device id)
7+
# PASSWORD - your account password
8+
#
9+
# Optional:
10+
# DEVICE_ID - device id if different from FMD_ID (used by Device tests)
11+
#
12+
# Example:
13+
BASE_URL=https://fmd.example.com
14+
FMD_ID=alice
15+
PASSWORD=secret
16+
#DEVICE_ID=alice-device

‎examples/tests/test_scripts/test_auth.py‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
"""
2+
Test: authenticate (FmdClient.create)
3+
Usage:
4+
From examples/tests: python test_scripts/test_auth.py
5+
From test_scripts: python test_auth.py
6+
"""
7+
import asyncio
8+
import sys
9+
from pathlib import Path
10+
11+
# Add parent directory to path so we can import utils
12+
sys.path.insert(0, str(Path(__file__).parent.parent))
13+
from utils.read_credentials import read_credentials
14+
15+
async def main():
16+
creds = read_credentials()
17+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
18+
print("Missing credentials. Copy credentials.txt.example -> credentials.txt and fill in BASE_URL, FMD_ID, PASSWORD")
19+
return
20+
from fmd_api import FmdClient
21+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
22+
print("Authenticated. access_token (first 12 chars):", (client.access_token or "")[:12])
23+
await client.close()
24+
25+
if __name__ == "__main__":
26+
asyncio.run(main())

‎examples/tests/test_scripts/test_commands.py‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
"""
2+
Test: send_command (ring, lock, camera, bluetooth)
3+
Usage:
4+
python test_scripts/test_commands.py <command>
5+
Examples:
6+
python test_scripts/test_commands.py ring
7+
python test_scripts/test_commands.py "camera front"
8+
python test_scripts/test_commands.py "bluetooth on"
9+
"""
10+
import asyncio
11+
import sys
12+
from pathlib import Path
13+
sys.path.insert(0, str(Path(__file__).parent.parent))
14+
from utils.read_credentials import read_credentials
15+
16+
async def main():
17+
creds = read_credentials()
18+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
19+
print("Missing credentials.")
20+
return
21+
if len(sys.argv) < 2:
22+
print("Usage: test_commands.py <command>")
23+
return
24+
cmd = sys.argv[1]
25+
from fmd_api import FmdClient
26+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
27+
try:
28+
ok = await client.send_command(cmd)
29+
print(f"Sent '{cmd}': {ok}")
30+
finally:
31+
await client.close()
32+
33+
if __name__ == "__main__":
34+
asyncio.run(main())

‎examples/tests/test_scripts/test_device.py‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
"""
2+
Test: Device class flows (refresh, get_location, fetch_pictures, download_photo)
3+
Usage:
4+
python test_scripts/test_device.py
5+
"""
6+
import asyncio
7+
import sys
8+
from pathlib import Path
9+
sys.path.insert(0, str(Path(__file__).parent.parent))
10+
from utils.read_credentials import read_credentials
11+
12+
async def main():
13+
creds = read_credentials()
14+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
15+
print("Missing credentials.")
16+
return
17+
device_id = creds.get("DEVICE_ID", creds.get("FMD_ID"))
18+
19+
from fmd_api import FmdClient
20+
from fmd_api.device import Device
21+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
22+
try:
23+
device = Device(client, device_id)
24+
print("Refreshing device (may return nothing if no data)...")
25+
await device.refresh()
26+
loc = await device.get_location()
27+
print("Cached location:", loc)
28+
# fetch pictures and attempt to download the first one
29+
pics = await device.fetch_pictures(5)
30+
print("Pictures listed:", len(pics))
31+
if pics:
32+
try:
33+
photo = await device.download_photo(pics[0])
34+
fn = "device_photo.jpg"
35+
with open(fn, "wb") as f:
36+
f.write(photo.data)
37+
print("Saved device photo to", fn)
38+
except Exception as e:
39+
print("Failed to download photo:", e)
40+
finally:
41+
await client.close()
42+
43+
if __name__ == "__main__":
44+
asyncio.run(main())

‎examples/tests/test_scripts/test_export.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
"""
2+
Test: export_data_zip (downloads export ZIP to provided filename)
3+
Usage:
4+
python test_scripts/test_export.py [output.zip]
5+
"""
6+
import asyncio
7+
import sys
8+
from pathlib import Path
9+
sys.path.insert(0, str(Path(__file__).parent.parent))
10+
from utils.read_credentials import read_credentials
11+
12+
async def main():
13+
creds = read_credentials()
14+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
15+
print("Missing credentials.")
16+
return
17+
out = sys.argv[1] if len(sys.argv) > 1 else "export_test.zip"
18+
from fmd_api import FmdClient
19+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
20+
try:
21+
await client.export_data_zip(out)
22+
print("Export saved to", out)
23+
finally:
24+
await client.close()
25+
26+
if __name__ == "__main__":
27+
asyncio.run(main())

‎examples/tests/test_scripts/test_locations.py‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
"""
2+
Test: get_locations + decrypt_data_blob
3+
Fetch most recent N blobs and decrypt each (prints parsed JSON).
4+
Usage:
5+
python test_scripts/test_locations.py [N]
6+
"""
7+
import asyncio
8+
import json
9+
import sys
10+
from pathlib import Path
11+
sys.path.insert(0, str(Path(__file__).parent.parent))
12+
from utils.read_credentials import read_credentials
13+
14+
async def main():
15+
creds = read_credentials()
16+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
17+
print("Missing credentials.")
18+
return
19+
num = -1
20+
if len(sys.argv) > 1:
21+
try:
22+
num = int(sys.argv[1])
23+
except:
24+
pass
25+
from fmd_api import FmdClient
26+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
27+
try:
28+
blobs = await client.get_locations(num_to_get=num if num != 0 else -1)
29+
print(f"Retrieved {len(blobs)} location blob(s)")
30+
for i, b in enumerate(blobs[:10]):
31+
try:
32+
dec = client.decrypt_data_blob(b)
33+
obj = json.loads(dec)
34+
print(f"Blob #{i}: {json.dumps(obj, indent=2)}")
35+
except Exception as e:
36+
print(f"Failed to decrypt/parse blob #{i}: {e}")
37+
finally:
38+
await client.close()
39+
40+
if __name__ == "__main__":
41+
asyncio.run(main())

‎examples/tests/test_scripts/test_pictures.py‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
"""
2+
Test: get_pictures and download/decrypt the first picture found
3+
Usage:
4+
python test_scripts/test_pictures.py
5+
"""
6+
import asyncio
7+
import base64
8+
import sys
9+
from pathlib import Path
10+
sys.path.insert(0, str(Path(__file__).parent.parent))
11+
from utils.read_credentials import read_credentials
12+
13+
async def main():
14+
creds = read_credentials()
15+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
16+
print("Missing credentials.")
17+
return
18+
19+
from fmd_api import FmdClient
20+
from fmd_api.device import Device
21+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
22+
try:
23+
pics = await client.get_pictures(10)
24+
print("Pictures returned:", len(pics))
25+
if not pics:
26+
print("No pictures available.")
27+
return
28+
29+
# Server sometimes returns list of dicts or list of base64 strings.
30+
# Try to extract a blob string:
31+
first = pics[0]
32+
blob = None
33+
if isinstance(first, dict):
34+
# try common keys
35+
for k in ("Data", "blob", "Blob", "data"):
36+
if k in first:
37+
blob = first[k]
38+
break
39+
# if picture metadata contains an encoded blob in a nested field, adjust as needed
40+
elif isinstance(first, str):
41+
blob = first
42+
43+
if not blob:
44+
print("Could not find picture blob inside first picture entry. Showing entry:")
45+
print(first)
46+
return
47+
48+
# decrypt to get inner base64 image string or bytes
49+
decrypted = client.decrypt_data_blob(blob)
50+
try:
51+
inner_b64 = decrypted.decode("utf-8").strip()
52+
img = base64.b64decode(inner_b64 + "=" * (-len(inner_b64) % 4))
53+
out = "picture_0.jpg"
54+
with open(out, "wb") as f:
55+
f.write(img)
56+
print("Saved picture to", out)
57+
except Exception as e:
58+
print("Decrypted payload not a base64 image string; saving raw bytes as picture_0.bin")
59+
with open("picture_0.bin", "wb") as f:
60+
f.write(decrypted)
61+
finally:
62+
await client.close()
63+
64+
if __name__ == "__main__":
65+
asyncio.run(main())

‎examples/tests/test_scripts/test_request_location.py‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
"""
2+
Test: request a new location command and then poll for the latest location.
3+
Usage:
4+
python test_scripts/test_request_location.py [provider] [wait_seconds]
5+
provider: one of all,gps,cell,last (default: all)
6+
wait_seconds: seconds to wait for the device to respond (default: 30)
7+
"""
8+
import asyncio
9+
import json
10+
import sys
11+
from pathlib import Path
12+
sys.path.insert(0, str(Path(__file__).parent.parent))
13+
from utils.read_credentials import read_credentials
14+
15+
async def main():
16+
creds = read_credentials()
17+
if not creds.get("BASE_URL") or not creds.get("FMD_ID") or not creds.get("PASSWORD"):
18+
print("Missing credentials.")
19+
return
20+
provider = sys.argv[1] if len(sys.argv) > 1 else "all"
21+
wait = int(sys.argv[2]) if len(sys.argv) > 2 else 30
22+
23+
from fmd_api import FmdClient
24+
client = await FmdClient.create(creds["BASE_URL"], creds["FMD_ID"], creds["PASSWORD"])
25+
try:
26+
ok = await client.request_location(provider)
27+
print("Request location sent:", ok)
28+
if ok and wait > 0:
29+
print(f"Waiting {wait} seconds for device to upload...")
30+
await asyncio.sleep(wait)
31+
blobs = await client.get_locations(5)
32+
print("Recent blobs:", len(blobs))
33+
if blobs:
34+
dec = client.decrypt_data_blob(blobs[0])
35+
print("Newest decrypted:", json.dumps(json.loads(dec), indent=2))
36+
finally:
37+
await client.close()
38+
39+
if __name__ == "__main__":
40+
asyncio.run(main())

‎examples/tests/utils/read_credentials.py‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
"""
2+
Utility: read credentials from credentials.txt (KEY=VALUE lines)
3+
4+
Place credentials.txt next to the test scripts (or set environment variables).
5+
"""
6+
from pathlib import Path
7+
import os
8+
9+
def read_credentials(path: str | Path = "credentials.txt") -> dict:
10+
"""Return dict of credentials from the given file. Falls back to env vars if not present."""
11+
creds = {}
12+
p = Path(path)
13+
if p.exists():
14+
for ln in p.read_text().splitlines():
15+
ln = ln.strip()
16+
if not ln or ln.startswith("#"):
17+
continue
18+
if "=" in ln:
19+
k, v = ln.split("=", 1)
20+
creds[k.strip()] = v.strip()
21+
# fallback to environment for keys not provided
22+
for k in ("BASE_URL", "FMD_ID", "PASSWORD", "DEVICE_ID"):
23+
if k not in creds and os.getenv(k):
24+
creds[k] = os.getenv(k)
25+
return creds

0 commit comments

Comments
 (0)