import sys import tempfile import time import unittest 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 seed_fixture import BABY_ID, build_seed_db from babytracker_api import ( DiaperStatus, FEED_KINDS, OpCode, Replica, decode_object, diaper, encode_object, format_date, formula, open_seed_db, read_cursors, sleep, ) NOW = time.time() BABY = {"objectID": BABY_ID, "name": "Sam", "newFlage": False} def wire(record): """Round-trip a record the way it travels in a transaction.""" return decode_object(encode_object(record)) def stamped(record, seconds): return {**record, "timestamp": format_date(datetime.fromtimestamp(NOW + seconds, timezone.utc))} class SeedTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.folder = tempfile.TemporaryDirectory() cls.connection = open_seed_db(build_seed_db(Path(cls.folder.name) / "seed.db", NOW)) @classmethod def tearDownClass(cls): cls.connection.close() cls.folder.cleanup() def setUp(self): self.replica = Replica.from_seed(self.connection) def details(self, entry_id): return next(e for e in self.replica.entries(BABY_ID) if e.id == entry_id).details def test_reads_babies_and_cursors(self): self.assertEqual(self.replica.babies[BABY_ID].name, "Sam") self.assertIs(self.replica.babies[BABY_ID].record["gender"], True) self.assertEqual(read_cursors(self.connection), {"phone": 3}) def test_entries_are_newest_first(self): times = [entry.time for entry in self.replica.entries(BABY_ID)] self.assertEqual(times, sorted(times, reverse=True)) self.assertEqual(self.replica.latest(BABY_ID).id, "d-new") def test_diaper_names_and_the_apps_dry_rule(self): self.assertEqual(self.details("d-new"), {"status": "poopy", "amount": "heavy"}) self.assertEqual(self.details("d-dry"), {"status": "dry", "amount": None}) def test_units_become_metric(self): self.assertAlmostEqual(self.details("f-oz")["volume_ml"], 118.294, places=2) self.assertEqual(self.details("f-ml")["volume_ml"], 100) self.assertAlmostEqual(self.details("t-1")["celsius"], 37.0, places=3) self.assertAlmostEqual(self.details("g-new")["weight_kg"], 6.0, places=3) def test_unset_growth_columns_are_none(self): self.assertIsNone(self.details("g-new")["length_cm"]) self.assertEqual(self.details("g-old")["length_cm"], 60) def test_selection_names_are_joined(self): self.assertEqual(self.details("m-1")["name"], "Vitamin D") def test_nursing_total_gives_the_end_time(self): entry = next(e for e in self.replica.entries(BABY_ID) if e.id == "n-1") self.assertEqual(entry.details["duration"], 12) self.assertEqual(entry.end - entry.time, timedelta(minutes=12)) self.assertEqual(entry.details["finish_side"], "right") self.assertEqual(entry.summary, "Nursing: left 5 min, right 7 min") def test_summaries(self): by_id = {entry.id: entry.summary for entry in self.replica.entries(BABY_ID)} self.assertEqual(by_id["d-new"], "Diaper: poopy, heavy") self.assertEqual(by_id["f-ml"], "Formula: 100 ml") self.assertEqual(by_id["s-1"], "Sleep: 45 min") self.assertEqual(by_id["m-1"], "Medication: Vitamin D") self.assertEqual(by_id["t-1"], "Temperature: 37.0 °C") self.assertEqual(by_id["k-1"], "Sick") def test_pump_belongs_to_no_baby(self): self.assertEqual([entry.id for entry in self.replica.entries(None)], ["p-1"]) def test_since_and_kind_filters(self): start = datetime.fromtimestamp(NOW - 4 * 3600, timezone.utc) self.assertEqual([e.id for e in self.replica.since(BABY_ID, start, {"Diaper"})], ["d-new", "d-dry"]) self.assertEqual({e.kind for e in self.replica.entries(BABY_ID, FEED_KINDS)}, {"Formula", "Nursing"}) def test_missing_tables_are_skipped(self): path = Path(self.folder.name) / "tiny.db" import sqlite3 connection = sqlite3.connect(path) connection.execute('CREATE TABLE "Baby" (ID, Timestamp, Name, DOB, DueDay, Gender, Picture)') connection.commit() connection.close() tiny = open_seed_db(path) self.addCleanup(tiny.close) self.assertEqual(Replica.from_seed(tiny).babies, {}) class LogTests(unittest.TestCase): def setUp(self): self.replica = Replica() self.replica.apply(OpCode.INSERT, {"BCObjectType": "Baby", **BABY}) def test_insert_update_delete(self): record = sleep(BABY, 30) self.replica.apply(OpCode.INSERT, stamped(wire(record), 0)) self.assertEqual(self.replica.latest(BABY_ID).details["duration"], 30) self.replica.apply(OpCode.UPDATE, stamped(wire({**record, "duration": 40}), 10)) self.assertEqual(self.replica.latest(BABY_ID).details["duration"], 40) self.replica.apply(OpCode.DELETE, stamped(wire(record), 20)) self.assertIsNone(self.replica.latest(BABY_ID)) def test_older_update_does_not_replace_newer(self): record = sleep(BABY, 30) self.replica.apply(OpCode.UPDATE, stamped(wire(record), 10)) self.replica.apply(OpCode.UPDATE, stamped(wire({**record, "duration": 99}), 5)) self.assertEqual(self.replica.latest(BABY_ID).details["duration"], 30) def test_older_update_does_not_undo_delete(self): record = sleep(BABY, 30) self.replica.apply(OpCode.DELETE, stamped(wire(record), 10)) self.replica.apply(OpCode.UPDATE, stamped(wire(record), 5)) self.assertIsNone(self.replica.latest(BABY_ID)) def test_enums_read_from_the_wire(self): self.replica.apply(OpCode.INSERT, wire(diaper(BABY, DiaperStatus.MIXED))) self.assertEqual(self.replica.latest(BABY_ID).details, {"status": "mixed", "amount": None}) def test_english_bottle(self): self.replica.apply(OpCode.INSERT, wire(formula(BABY, 1, english=True))) self.assertAlmostEqual(self.replica.latest(BABY_ID).details["volume_ml"], 29.5735) def test_localized_date_is_understood(self): record = {**wire(sleep(BABY, 1)), "time": "2026-01-02 03:00:05 nachm. +0000"} self.replica.apply(OpCode.INSERT, record) self.assertEqual(self.replica.latest(BABY_ID).time.hour, 15) def test_unknown_type_with_a_time_is_kept(self): record = {"BCObjectType": "FutureThing", "objectID": "x", "baby": BABY, "time": "2026-01-02 03:00:05 +0000"} self.replica.apply(OpCode.INSERT, record) self.assertEqual(self.replica.latest(BABY_ID).summary, "FutureThing") def test_selection_and_picture_records_are_ignored(self): self.replica.apply(OpCode.INSERT, {"BCObjectType": "OtherFeedSelection", "objectID": "s", "name": "Fruit"}) self.replica.apply(OpCode.INSERT, {"BCObjectType": "Picture", "objectID": "p", "fileName": "p"}) self.assertIsNone(self.replica.latest(BABY_ID)) def test_baby_added_renamed_and_deleted(self): other = {"BCObjectType": "Baby", "objectID": "b2", "name": "Alex"} self.replica.apply(OpCode.INSERT, other) self.replica.apply(OpCode.UPDATE, {**other, "name": "Alexa"}) self.assertEqual(self.replica.babies["b2"].name, "Alexa") self.replica.apply(OpCode.DELETE, other) self.assertNotIn("b2", self.replica.babies) def test_conflict_moves_entries_to_the_winner(self): loser = {"BCObjectType": "Baby", "objectID": "b2", "name": "Sam"} self.replica.apply(OpCode.INSERT, loser) self.replica.apply(OpCode.INSERT, wire(sleep({"objectID": "b2"}, 15))) winner = {"BCObjectType": "Baby", **BABY} self.replica.apply(OpCode.CONFLICT, {"BCObjectType": "ConflictRecord", "objectID": "c", "winner": winner, "loser": loser}) self.assertNotIn("b2", self.replica.babies) self.assertIn(BABY_ID, self.replica.babies) self.assertEqual(self.replica.latest(BABY_ID).details["duration"], 15) self.assertIsNone(self.replica.latest("b2")) def test_selection_conflict_changes_nothing(self): selection = {"BCObjectType": "OtherFeedSelection", "objectID": "s"} self.replica.apply(OpCode.CONFLICT, {"BCObjectType": "ConflictRecord", "winner": selection, "loser": selection}) self.assertEqual(list(self.replica.babies), [BABY_ID]) if __name__ == "__main__": unittest.main()