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()