diff --git a/agent-office/app/node_monitor.py b/agent-office/app/node_monitor.py index bc53673..8bbd9a6 100644 --- a/agent-office/app/node_monitor.py +++ b/agent-office/app/node_monitor.py @@ -28,9 +28,13 @@ def _get_redis(): return _redis -def _beat_age(ts_str, now): +def _beat_age(ts, now): + """ts는 ISO-8601 문자열 또는 epoch 숫자(매물알림 스펙 §4.4) 둘 다 허용.""" try: - beat = dt.datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + if isinstance(ts, (int, float)): + beat = dt.datetime.fromtimestamp(ts, tz=dt.timezone.utc) + else: + beat = dt.datetime.fromisoformat(ts.replace("Z", "+00:00")) return max(0, int((now - beat).total_seconds())) except Exception: return None diff --git a/agent-office/tests/test_node_monitor.py b/agent-office/tests/test_node_monitor.py index cf7115c..509fea7 100644 --- a/agent-office/tests/test_node_monitor.py +++ b/agent-office/tests/test_node_monitor.py @@ -224,3 +224,25 @@ async def test_naver_fetch_http_pull_link(): st = await node_monitor.collect_status(redis=r) link = next(l for l in st["links"] if l["from"] == "naver-fetch") assert link["type"] == "http-pull" and link["to"] == "nas-realestate" and link["status"] == "healthy" + + +# ── ts epoch 포맷 지원 (매물알림 스펙 §4.4: naver-fetch는 ts를 epoch 정수로 발신) ── + +@pytest.mark.asyncio +async def test_epoch_ts_fresh_is_alive(): + """ts가 epoch 정수(스펙 §4.4)여도 신선하면 alive=True.""" + epoch_now = int(dt.datetime.now(dt.timezone.utc).timestamp()) + r = FakeRedis(kv={"worker:naver-fetch:heartbeat": _hb("naver-fetch", "fetcher", "idle", ts=epoch_now)}) + st = await node_monitor.collect_status(redis=r) + nf = next(w for w in st["workers"] if w["name"] == "naver-fetch") + assert nf["alive"] is True and nf["last_beat_age_s"] is not None + + +@pytest.mark.asyncio +async def test_epoch_ts_stale_is_dead(): + """epoch ts가 90s 초과 과거면 staleness 판정으로 alive=False.""" + epoch_old = int(dt.datetime.now(dt.timezone.utc).timestamp()) - 300 + r = FakeRedis(kv={"worker:naver-fetch:heartbeat": _hb("naver-fetch", "fetcher", "idle", ts=epoch_old)}) + st = await node_monitor.collect_status(redis=r) + nf = next(w for w in st["workers"] if w["name"] == "naver-fetch") + assert nf["alive"] is False and nf["last_beat_age_s"] >= 300