arneandClaude Opus 5.5
6466f002e0
Python client for the Baby Tracker sync server, reverse engineered from the Android app (4.53), and a HACS-installable Home Assistant integration built on it. Each baby gets a log calendar and summary sensors, and actions add entries that sync back to the group. Reading has been run against the real server. Uploading and running inside real Home Assistant have not. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
393 lines
16 KiB
Python
393 lines
16 KiB
Python
import base64
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
import load_library # noqa: F401
|
|
|
|
from mock_server import MockServer
|
|
|
|
from babytracker_api import (
|
|
AccountChangedError,
|
|
AccountStateError,
|
|
AuthenticationError,
|
|
BabyTrackerClient,
|
|
ConflictError,
|
|
DiaperAmount,
|
|
DiaperStatus,
|
|
MissingSeedError,
|
|
NotFoundError,
|
|
NursingFinishSide,
|
|
OpCode,
|
|
Transaction,
|
|
decode_object,
|
|
diaper,
|
|
encode_object,
|
|
extract_seed_db,
|
|
format_date,
|
|
formula,
|
|
growth,
|
|
new_activity,
|
|
nursing,
|
|
open_seed_db,
|
|
parse_date,
|
|
read_babies,
|
|
read_cursors,
|
|
sleep,
|
|
)
|
|
|
|
BABY = {"objectID": "baby-1", "name": "Sam", "gender": True, "newFlage": False}
|
|
|
|
|
|
def log_entry(sync_id, op="0"):
|
|
payload = encode_object({"BCObjectType": "Bath", "objectID": f"o{sync_id}"})
|
|
return {"SyncID": sync_id, "OPCode": op, "Transaction": payload}
|
|
|
|
|
|
class ServerTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.server = MockServer()
|
|
self.server.__enter__()
|
|
self.addCleanup(self.server.__exit__)
|
|
self.state = self.server.state
|
|
self.client = BabyTrackerClient(
|
|
self.state.email, self.state.password, device_uuid="me", device_name="Test", base_url=self.server.url
|
|
)
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.tmp = Path(tmp.name)
|
|
|
|
|
|
class SessionTests(ServerTestCase):
|
|
def test_login_sends_the_apps_body_and_returns_account_id(self):
|
|
self.assertEqual(self.client.login(), 42)
|
|
self.assertEqual(
|
|
self.state.login_bodies[0],
|
|
{
|
|
"AppInfo": {"AccountType": 0, "AppType": 0},
|
|
"Device": {"DeviceName": "Test", "DeviceOSInfo": "Python", "DeviceUUID": "me"},
|
|
"EmailAddress": "parent@example.com",
|
|
"Password": "correct-horse",
|
|
},
|
|
)
|
|
|
|
def test_bad_password_raises(self):
|
|
self.client.password = "wrong"
|
|
with self.assertRaises(AuthenticationError) as caught:
|
|
self.client.login()
|
|
self.assertEqual(caught.exception.status, 401)
|
|
|
|
def test_login_307_is_not_followed(self):
|
|
self.state.login_status = 307
|
|
with self.assertRaises(AccountStateError):
|
|
self.client.login()
|
|
self.assertNotIn(("POST", "/redirected"), self.state.requests)
|
|
|
|
def test_expired_session_logs_in_again_and_retries(self):
|
|
self.client.login()
|
|
self.state.valid_sessions = False
|
|
self.assertEqual(self.client.get_devices(), [])
|
|
self.assertEqual(len(self.state.login_bodies), 2)
|
|
|
|
def test_first_request_without_login_works(self):
|
|
self.assertEqual(self.client.get_devices(), [])
|
|
|
|
def test_account_change_on_relogin_raises(self):
|
|
self.client.login()
|
|
self.state.valid_sessions = False
|
|
self.state.account_id = 99
|
|
with self.assertRaises(AccountChangedError):
|
|
self.client.get_devices()
|
|
|
|
def test_logout_and_push_token(self):
|
|
self.client.login()
|
|
self.client.register_push_token("tok")
|
|
self.assertEqual(json.loads(self.state.multipart["token"][1]), {"TokenNative": "tok", "TokenAppType": 5})
|
|
self.client.logout()
|
|
self.assertIn(("DELETE", "/session"), self.state.requests)
|
|
|
|
|
|
class PullTests(ServerTestCase):
|
|
def test_pages_until_a_short_page(self):
|
|
self.state.logs["other"] = [log_entry(i) for i in range(1, 1203)]
|
|
self.client.login()
|
|
items = list(self.client.iter_transactions("other"))
|
|
self.assertEqual([item.sync_id for item in items], list(range(1, 1203)))
|
|
paths = [path for _, path in self.state.requests if "transaction" in path]
|
|
self.assertEqual(
|
|
paths,
|
|
["/account/transaction/other/0", "/account/transaction/other/500", "/account/transaction/other/1000"],
|
|
)
|
|
|
|
def test_exact_page_boundary_fetches_one_empty_page(self):
|
|
self.state.logs["other"] = [log_entry(i) for i in range(1, 501)]
|
|
self.assertEqual(len(list(self.client.iter_transactions("other"))), 500)
|
|
|
|
def test_server_ignoring_cursor_does_not_loop_or_repeat(self):
|
|
real = self.client.get_transactions
|
|
full_page = [Transaction.from_json(log_entry(i)) for i in range(1, 501)]
|
|
self.client.get_transactions = lambda device, after=0: full_page
|
|
self.addCleanup(setattr, self.client, "get_transactions", real)
|
|
self.assertEqual(len(list(self.client.iter_transactions("other"))), 500)
|
|
|
|
def test_pull_skips_own_device_and_seen_items_and_tracks_cursors(self):
|
|
self.state.devices = [
|
|
{"DeviceUUID": "me", "LastSyncID": 5},
|
|
{"DeviceUUID": "a", "LastSyncID": 3},
|
|
{"DeviceUUID": "b", "LastSyncID": 2},
|
|
]
|
|
self.state.logs = {
|
|
"me": [log_entry(5)],
|
|
"a": [log_entry(1), log_entry(2), log_entry(3)],
|
|
"b": [log_entry(1), log_entry(2)],
|
|
}
|
|
cursors = {"a": 1, "b": 2}
|
|
pulled = [(device, item.sync_id) for device, item in self.client.pull(cursors)]
|
|
self.assertEqual(pulled, [("a", 2), ("a", 3)])
|
|
self.assertEqual(cursors, {"a": 3, "b": 2})
|
|
self.assertEqual(list(self.client.pull(cursors)), [])
|
|
|
|
def test_opcode_accepts_numbers_as_well_as_strings(self):
|
|
self.state.logs["a"] = [log_entry(1, op=2), log_entry(2, op="1")]
|
|
items = self.client.get_transactions("a")
|
|
self.assertEqual([item.op for item in items], [OpCode.DELETE, OpCode.UPDATE])
|
|
self.assertEqual(items[0].object_type, "Bath")
|
|
|
|
|
|
class PushTests(ServerTestCase):
|
|
def test_add_uses_gson_wire_format(self):
|
|
when = datetime(2026, 11, 4, 8, 30, tzinfo=timezone.utc)
|
|
record = diaper(BABY, DiaperStatus.POOPY, DiaperAmount.HEAVY, time=when, note="x")
|
|
self.client.add(record)
|
|
|
|
upload = self.state.uploads[0]
|
|
self.assertEqual(set(upload), {"SyncID", "OPCode", "Transaction"})
|
|
self.assertEqual(upload["SyncID"], 1)
|
|
self.assertEqual(upload["OPCode"], "0")
|
|
|
|
body = json.loads(base64.b64decode(upload["Transaction"]))
|
|
self.assertEqual(body["BCObjectType"], "Diaper")
|
|
self.assertEqual(body["status"], "1")
|
|
self.assertEqual(body["amount"], "3")
|
|
self.assertEqual(body["time"], "2026-11-04 08:30:00 +0000")
|
|
self.assertEqual(body["baby"], BABY)
|
|
self.assertNotIn("BCObjectType", body["baby"])
|
|
self.assertIs(body["pictureLoaded"], True)
|
|
self.assertEqual(body["pictureNote"], [])
|
|
|
|
def test_sync_ids_resume_from_server_then_count(self):
|
|
self.state.devices = [{"DeviceUUID": "me", "LastSyncID": 7}, {"DeviceUUID": "x", "LastSyncID": 100}]
|
|
self.client.login()
|
|
for _ in range(3):
|
|
self.client.add(sleep(BABY, 30))
|
|
self.assertEqual([u["SyncID"] for u in self.state.uploads], [8, 9, 10])
|
|
device_lists = [r for r in self.state.requests if r == ("GET", "/account/device")]
|
|
self.assertEqual(len(device_lists), 1)
|
|
|
|
def test_failed_upload_does_not_consume_a_sync_id(self):
|
|
self.client.login()
|
|
record = sleep(BABY, 30)
|
|
original = self.client.upload_transaction
|
|
self.client.upload_transaction = lambda item: (_ for _ in ()).throw(ConflictError(409))
|
|
with self.assertRaises(ConflictError):
|
|
self.client.add(record)
|
|
self.client.upload_transaction = original
|
|
self.client.add(record)
|
|
self.assertEqual(self.state.uploads[0]["SyncID"], 1)
|
|
|
|
def test_update_and_delete_send_whole_record_with_new_timestamp(self):
|
|
old = datetime(2020, 1, 1, tzinfo=timezone.utc)
|
|
record = {**sleep(BABY, 30), "timestamp": old}
|
|
self.client.update(record)
|
|
self.client.delete(record)
|
|
self.assertEqual([u["OPCode"] for u in self.state.uploads], ["1", "2"])
|
|
for upload in self.state.uploads:
|
|
body = json.loads(base64.b64decode(upload["Transaction"]))
|
|
self.assertEqual(body["objectID"], record["objectID"])
|
|
self.assertGreater(parse_date(body["timestamp"]), old)
|
|
self.assertEqual(record["timestamp"], old)
|
|
|
|
|
|
class FileTests(ServerTestCase):
|
|
def _make_seed(self):
|
|
db_path = self.tmp / "source.db"
|
|
connection = sqlite3.connect(db_path)
|
|
connection.execute(
|
|
"CREATE TABLE Baby (ID TEXT, Timestamp REAL, Name TEXT, DOB REAL, DueDay REAL, Gender INT, Picture TEXT)"
|
|
)
|
|
connection.execute('CREATE TABLE MergedTransaction ("DeviceID" TEXT, "SyncID" INTEGER)')
|
|
connection.execute("INSERT INTO Baby VALUES ('b1', 1700000000.5, 'Sam', 1600000000, 0, 1, NULL)")
|
|
connection.execute("INSERT INTO MergedTransaction VALUES ('a', 9)")
|
|
connection.commit()
|
|
connection.close()
|
|
zip_path = self.tmp / "made.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.write(db_path, "EasyLog.db")
|
|
archive.writestr("../../evil.txt", "ignored")
|
|
return zip_path.read_bytes()
|
|
|
|
def test_seed_download_and_read(self):
|
|
self.state.seed = self._make_seed()
|
|
seed_zip = self.client.download_seed(self.tmp / "seed.zip")
|
|
connection = open_seed_db(extract_seed_db(seed_zip, self.tmp / "seed.db"))
|
|
self.addCleanup(connection.close)
|
|
|
|
babies = read_babies(connection)
|
|
self.assertEqual(babies[0]["objectID"], "b1")
|
|
self.assertIs(babies[0]["gender"], True)
|
|
self.assertEqual(babies[0]["dob"], "2020-09-13 12:26:40 +0000")
|
|
self.assertNotIn("dueDay", babies[0])
|
|
self.assertNotIn("pictureName", babies[0])
|
|
self.assertEqual(read_cursors(connection), {"a": 9})
|
|
with self.assertRaises(sqlite3.OperationalError):
|
|
connection.execute("DELETE FROM Baby")
|
|
self.assertFalse((self.tmp.parent / "evil.txt").exists())
|
|
|
|
self.client.add(diaper(babies[0], DiaperStatus.WET))
|
|
body = decode_object(self.state.uploads[0]["Transaction"])
|
|
self.assertEqual(body["baby"]["objectID"], "b1")
|
|
|
|
def test_missing_seed_raises(self):
|
|
with self.assertRaises(NotFoundError):
|
|
self.client.download_seed(self.tmp / "seed.zip")
|
|
|
|
def test_seed_zip_without_database_raises(self):
|
|
path = self.tmp / "empty.zip"
|
|
with zipfile.ZipFile(path, "w") as archive:
|
|
archive.writestr("other.txt", "x")
|
|
with self.assertRaises(ValueError):
|
|
extract_seed_db(path, self.tmp / "seed.db")
|
|
|
|
def test_picture_upload_matches_the_apps_bytes(self):
|
|
path = self.tmp / "abc.jpg"
|
|
path.write_bytes(b"\xff\xd8jpeg")
|
|
self.client.login()
|
|
self.client.upload_picture(path)
|
|
content_type, body = self.state.multipart["/account/picture"]
|
|
self.assertEqual(content_type, "multipart/form-data; boundary=__BABY_TRACKER_BOUNDARY____Photo__")
|
|
self.assertEqual(
|
|
body,
|
|
b"\r\n--__BABY_TRACKER_BOUNDARY____Photo__\r\n"
|
|
b'Content-Disposition: form-data; name="picture"; filename="abc.jpg"\r\n'
|
|
b"Content-Type: image/jpeg\r\n\r\n"
|
|
b"\xff\xd8jpeg"
|
|
b"\r\n--__BABY_TRACKER_BOUNDARY____Photo__--\r\n",
|
|
)
|
|
|
|
def test_picture_download_quotes_the_name(self):
|
|
self.state.files["abc__thumbnail.jpg"] = b"data"
|
|
self.client.login()
|
|
path = self.client.download_picture("abc__thumbnail.jpg", self.tmp / "pic.jpg")
|
|
self.assertEqual(path.read_bytes(), b"data")
|
|
with self.assertRaises(NotFoundError):
|
|
self.client.download_picture("../session", self.tmp / "x.jpg")
|
|
self.assertIn(("GET", "/account/picture/..%2Fsession"), self.state.requests)
|
|
|
|
|
|
class GroupTests(ServerTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.seed = self.tmp / "local.zip"
|
|
self.seed.write_bytes(b"zipdata")
|
|
self.picture = self.tmp / "p.jpg"
|
|
self.picture.write_bytes(b"jpg")
|
|
|
|
def test_create_group_runs_the_apps_steps_in_order(self):
|
|
self.assertEqual(self.client.create_group(self.seed, [self.picture]), 42)
|
|
self.assertEqual(
|
|
self.state.requests,
|
|
[
|
|
("POST", "/staging_account"),
|
|
("POST", "/staging_account/seed"),
|
|
("POST", "/staging_account/picture"),
|
|
("PUT", "/staging_account"),
|
|
],
|
|
)
|
|
content_type, body = self.state.multipart["/staging_account/seed"]
|
|
self.assertEqual(content_type, "multipart/form-data; boundary=__BABY_TRACKER_BOUNDARY____SEED__")
|
|
self.assertIn(b'name="seed"; filename="seed.zip"\r\nContent-Type: application/zip\r\n\r\nzipdata', body)
|
|
|
|
def test_create_group_for_taken_email_raises(self):
|
|
self.client.email = "taken@example.com"
|
|
with self.assertRaises(ConflictError):
|
|
self.client.create_group(self.seed)
|
|
|
|
def test_reset_requires_confirmation(self):
|
|
with self.assertRaises(ValueError):
|
|
self.client.reset_group(self.seed)
|
|
self.assertEqual(self.state.requests, [])
|
|
|
|
def test_reset_with_confirmation(self):
|
|
self.assertEqual(self.client.reset_group(self.seed, confirm=True), 42)
|
|
self.assertEqual(self.state.requests[0], ("POST", "/account_reset"))
|
|
|
|
def test_finalizing_without_seed_raises(self):
|
|
self.client.login()
|
|
with self.assertRaises(MissingSeedError):
|
|
self.client._check(self.client._send("PUT", "/staging_account"), 200)
|
|
|
|
|
|
class ModelTests(unittest.TestCase):
|
|
def test_date_round_trip_converts_to_gmt(self):
|
|
tz = timezone(timedelta(hours=2))
|
|
text = format_date(datetime(2026, 11, 4, 10, 0, 5, tzinfo=tz))
|
|
self.assertEqual(text, "2026-11-04 08:00:05 +0000")
|
|
self.assertEqual(parse_date(text), datetime(2026, 11, 4, 8, 0, 5, tzinfo=timezone.utc))
|
|
|
|
def test_naive_datetime_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
format_date(datetime(2026, 1, 1))
|
|
|
|
def test_parses_the_apps_localized_fallback_dates(self):
|
|
expected = datetime(2026, 11, 4, 15, 0, 5, tzinfo=timezone.utc)
|
|
for text in (
|
|
"2026-11-04 03:00:05 PM +0000",
|
|
"2026-11-04 03:00:05 p.m. +0000",
|
|
"2026-11-04 03:00:05 nachm. +0000",
|
|
"2026-11-04 03:00:05 午後 +0000",
|
|
"2026-11-04 03:00:05 下午 +0000",
|
|
"2026-11-04 03:00:05 오후 +0000",
|
|
"2026-11-04 03:00:05 em +0000",
|
|
):
|
|
self.assertEqual(parse_date(text), expected, text)
|
|
|
|
def test_unknown_fields_survive_a_round_trip(self):
|
|
record = {"BCObjectType": "FutureType", "newThing": {"a": [1, "é"]}}
|
|
self.assertEqual(decode_object(encode_object(record)), record)
|
|
|
|
def test_record_needs_a_type(self):
|
|
with self.assertRaises(ValueError):
|
|
encode_object({"objectID": "x"})
|
|
|
|
def test_enums_become_strings_even_when_nested(self):
|
|
record = new_activity("Pump", None, extra=[{"sides": DiaperStatus.MIXED}])
|
|
body = decode_object(encode_object(record))
|
|
self.assertEqual(body["extra"], [{"sides": "2"}])
|
|
self.assertNotIn("baby", body)
|
|
|
|
def test_builders(self):
|
|
body = decode_object(encode_object(formula(BABY, 4, english=True)))
|
|
self.assertEqual(body["amount"], {"englishMeasure": True, "value": 4.0})
|
|
body = decode_object(encode_object(nursing(BABY, 5, 7, finish_side=NursingFinishSide.RIGHT)))
|
|
self.assertEqual((body["leftDuration"], body["rightDuration"], body["finishSide"]), (5, 7, "2"))
|
|
body = decode_object(encode_object(growth(BABY, weight=3.5)))
|
|
self.assertEqual(body["weight"], {"englishMeasure": False, "value": 3.5})
|
|
self.assertNotIn("length", body)
|
|
self.assertEqual(decode_object(encode_object(diaper(BABY, DiaperStatus.WET)))["amount"], "5")
|
|
|
|
def test_builder_time_and_note(self):
|
|
when = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
|
|
body = decode_object(encode_object(sleep(BABY, 90, time=when, note="nap")))
|
|
self.assertEqual((body["time"], body["note"], body["duration"]), ("2026-01-02 03:04:05 +0000", "nap", 90))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|