turnstone/tests/test_services_diagnose.py
pyr0ball 828b69768a refactor: rename ingest → glean throughout codebase
Renames the app/ingest/ package to app/glean/ and updates all
references across Python modules, shell scripts, Vue components,
tests, and documentation.

Intentionally preserved:
- SQLite column name ingest_time (avoids schema migration)
- RetrievedEntry.ingest_time field (maps to the column above)
- Any public-facing JSON keys that reference ingest_time

Changes by category:
- app/ingest/ → app/glean/ (full package move, all parsers)
- app/tasks/ingest_scheduler.py → app/tasks/glean_scheduler.py
- scripts/ingest_corpus.py → scripts/glean_corpus.py
- tests/test_ingest_*.py → tests/test_glean_*.py
- Docstrings, log messages, comments: ingest → glean
- Env var: TURNSTONE_INGEST_INTERVAL → TURNSTONE_GLEAN_INTERVAL
- Shell scripts: glean.log, glean_corpus.py references
- README.md: multi-source ingest → multi-source glean
- .env.example: updated env var name
- patterns/: new diagnostic patterns from 2026-05-20 SSH incident
  (service_crash_loop, pkg_daemon_restart, ssh_forward_conflict)
- SourcesView.vue: pipeline label updated
- All test import paths updated to app.glean.*

285 tests passing.
2026-05-20 23:02:55 -07:00

63 lines
2.8 KiB
Python

"""Tests for app/services/diagnose.py — parse_time_window."""
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import patch
from app.services.diagnose import diagnose, parse_time_window
def test_no_time_phrase_falls_back_to_last_60_min():
with patch("app.services.diagnose._HAS_DATEPARSER", True), \
patch("app.services.diagnose._search_dates", return_value=None):
since, until, keywords = parse_time_window("plex stopped playing audio")
assert since is not None and until is not None
assert keywords == "plex stopped playing audio"
diff = (datetime.fromisoformat(until) - datetime.fromisoformat(since)).total_seconds()
assert abs(diff - 3600) < 5
def test_time_phrase_detected_produces_60min_window():
dt = datetime(2026, 5, 11, 14, 0, tzinfo=timezone.utc)
with patch("app.services.diagnose._HAS_DATEPARSER", True), \
patch("app.services.diagnose._search_dates", return_value=[("around 2pm", dt)]):
since, until, keywords = parse_time_window("plex stopped audio around 2pm")
diff = (datetime.fromisoformat(until) - datetime.fromisoformat(since)).total_seconds()
assert abs(diff - 3600) < 5
def test_time_phrase_stripped_from_keywords():
dt = datetime(2026, 5, 11, 14, 0, tzinfo=timezone.utc)
with patch("app.services.diagnose._HAS_DATEPARSER", True), \
patch("app.services.diagnose._search_dates", return_value=[("around 2pm", dt)]):
_, _, keywords = parse_time_window("plex stopped audio around 2pm")
assert "around 2pm" not in keywords
assert "plex" in keywords
def test_no_dateparser_falls_back_to_60min():
with patch("app.services.diagnose._HAS_DATEPARSER", False), \
patch("app.services.diagnose._search_dates", None):
since, until, keywords = parse_time_window("plex stopped playing audio")
assert keywords == "plex stopped playing audio"
diff = (datetime.fromisoformat(until) - datetime.fromisoformat(since)).total_seconds()
assert abs(diff - 3600) < 5
def test_keywords_cleaned_of_extra_spaces():
dt = datetime(2026, 5, 11, 14, 0, tzinfo=timezone.utc)
with patch("app.services.diagnose._HAS_DATEPARSER", True), \
patch("app.services.diagnose._search_dates", return_value=[("around 2pm", dt)]):
_, _, keywords = parse_time_window("plex stopped audio around 2pm extra")
assert " " not in keywords
def test_diagnose_with_explicit_window_sets_time_detected(tmp_path):
from app.glean.pipeline import ensure_schema
db = tmp_path / "test.db"
ensure_schema(db)
result = diagnose(db, query="plex", since="2026-05-11T14:00:00+00:00", until="2026-05-11T15:00:00+00:00")
assert result["summary"]["time_detected"] is True
assert result["summary"]["total"] == 0 # empty DB
assert result["summary"]["window_start"] == "2026-05-11T14:00:00+00:00"