ha-babytracker/tests/test_replica.py

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