Spaces:
Sleeping
Sleeping
| """Autonomous mode pushes to Google Calendar ONLY after dedup. | |
| Locks in the fix for the duplicate-push bug: the pipeline used to run with | |
| push_gcal=True (pushing ALL plan events) and dedup only afterwards — so every | |
| rolling-window re-run re-pushed already-captured events, directly contradicting | |
| dedup.py's "don't create the same event twice".""" | |
| import app as app_module | |
| import calendar_out.gcal as gcal | |
| from server import dedup | |
| def _seed_feed(tmp_path, monkeypatch): | |
| monkeypatch.setattr(app_module, "FEED_PATH", tmp_path / "feed.json") | |
| monkeypatch.setenv("DEDUP_PATH", str(tmp_path / "seen.json")) | |
| dedup.reset() | |
| app_module._append_feed([{ | |
| "chat": "Fam", "sender": "Mom", "text": "dinner 6pm tomorrow", | |
| "timestamp": "2026-06-05T10:00:00", "images": [], "is_from_me": False, | |
| }]) | |
| def test_pipeline_runs_without_push_and_only_fresh_events_are_pushed(monkeypatch, tmp_path): | |
| _seed_feed(tmp_path, monkeypatch) | |
| pushes: list[list[str]] = [] | |
| monkeypatch.setattr( | |
| gcal, "push_events", | |
| lambda evs, calendar_id="primary": pushes.append([e.title for e in evs]) or [""] * len(evs), | |
| ) | |
| reqs = [] | |
| real_pipeline = app_module.run_pipeline | |
| def spy_pipeline(req): | |
| reqs.append(req) | |
| return real_pipeline(req) | |
| monkeypatch.setattr(app_module, "run_pipeline", spy_pipeline) | |
| app_module._run_autonomous({"Fam"}) | |
| # the pipeline itself must NOT push — push happens after dedup, on the | |
| # filtered set only | |
| assert reqs and all(r.push_gcal is False for r in reqs) | |
| assert len(pushes) == 1 and pushes[0], "fresh events pushed exactly once" | |
| # re-run over the same rolling window: nothing fresh -> NO second push | |
| app_module._run_autonomous({"Fam"}) | |
| assert len(pushes) == 1, "already-captured events must never be re-pushed" | |
| def test_push_failure_is_retried_not_swallowed(monkeypatch, tmp_path): | |
| """A transient push failure must NOT mark events seen — they'd be filtered | |
| out forever ('seen' but never delivered). The next run retries them.""" | |
| _seed_feed(tmp_path, monkeypatch) | |
| def boom(evs, calendar_id="primary"): | |
| raise RuntimeError("transient gcal outage") | |
| monkeypatch.setattr(gcal, "push_events", boom) | |
| app_module._run_autonomous({"Fam"}) # must not raise | |
| assert dedup._load() == [], "failed push must not mark events seen" | |
| # outage over: the SAME events get delivered on the next run | |
| pushes = [] | |
| monkeypatch.setattr( | |
| gcal, "push_events", | |
| lambda evs, calendar_id="primary": pushes.append(len(evs)) or [""] * len(evs), | |
| ) | |
| app_module._run_autonomous({"Fam"}) | |
| assert pushes == [1] | |
| assert len(dedup._load()) >= 1, "successful push marks events seen" | |