feat(cli): tell people at a terminal about a newer release and Tauric's notices

- once a day a background thread records the latest GitHub release and the notices for each version; the welcome screen shows them from that cache, so no run waits on the network
- the release notice names the upgrade command for how TradingAgents was installed
- TRADINGAGENTS_NO_UPDATE_CHECK=1 turns the check off; unattended and CI runs never make it
- notices are read from a new `notices` key, so earlier versions keep reading `announcements`
This commit is contained in:
Yijia-Xiao
2026-10-16 21:11:55 +00:00
parent c9be8b6047
commit c3cdaeb4eb
10 changed files with 574 additions and 91 deletions
+3 -1
View File
@@ -12,8 +12,10 @@ RUN pip install --no-cache-dir .
FROM python:3.13-slim
# TRADINGAGENTS_IN_DOCKER lets an update notice name the rebuild command.
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
PYTHONUNBUFFERED=1 \
TRADINGAGENTS_IN_DOCKER=1
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
-53
View File
@@ -1,53 +0,0 @@
import getpass
import sys
import requests
from rich.console import Console
from rich.panel import Panel
from cli.config import CLI_CONFIG
def fetch_announcements(url: str = None, timeout: float = None) -> dict:
"""Fetch announcements from endpoint. Returns dict with announcements and settings."""
endpoint = url or CLI_CONFIG["announcements_url"]
timeout = timeout or CLI_CONFIG["announcements_timeout"]
fallback = CLI_CONFIG["announcements_fallback"]
try:
response = requests.get(endpoint, timeout=timeout)
response.raise_for_status()
data = response.json()
return {
"announcements": data.get("announcements", [fallback]),
"require_attention": data.get("require_attention", False),
}
except Exception:
return {
"announcements": [fallback],
"require_attention": False,
}
def display_announcements(console: Console, data: dict) -> None:
"""Display announcements panel. Prompts for Enter if require_attention is True."""
announcements = data.get("announcements", [])
require_attention = data.get("require_attention", False)
if not announcements:
return
content = "\n".join(announcements)
panel = Panel(
content,
border_style="cyan",
padding=(1, 2),
title="Announcements",
)
console.print(panel)
if require_attention and sys.stdin.isatty():
getpass.getpass("Press Enter to continue...")
else:
console.print()
-6
View File
@@ -1,6 +0,0 @@
CLI_CONFIG = {
# Announcements
"announcements_url": "https://api.tauric.ai/v1/announcements",
"announcements_timeout": 1.0,
"announcements_fallback": "[cyan]For more information, please visit[/cyan] [link=https://github.com/TauricResearch]https://github.com/TauricResearch[/link]",
}
+6 -4
View File
@@ -8,7 +8,7 @@ import typer
from rich.align import Align
from rich.panel import Panel
from cli.announcements import display_announcements, fetch_announcements
from cli import updates
from cli.display import (
console,
)
@@ -124,9 +124,11 @@ def _prompt_selections(prefs, flags):
console.print()
console.print() # Add vertical space before announcements
# Fetch and display announcements (silent on failure)
announcements = fetch_announcements()
display_announcements(console, announcements)
# What's new comes from a cache refreshed in the background, and only for
# someone at a terminal: a run never waits on it and unattended runs skip it.
if updates.enabled():
updates.show(console)
updates.start_background_refresh()
def create_question_box(title, prompt, default=None):
box_content = f"[bold]{title}[/bold]\n"
+220
View File
@@ -0,0 +1,220 @@
"""What's new, for the person at the terminal: a newer release, and notices from Tauric Research.
At most once a day a background thread asks GitHub for the latest release and
the announcements server for notices, and writes the answers to a small cache.
The lines shown at start-up are read from that cache, so a run never waits on
the network and news shows up on the next run. Nothing identifying is sent.
Unattended, scripted and CI runs contact no server; set
TRADINGAGENTS_NO_UPDATE_CHECK=1 to turn the check off.
"""
import importlib.metadata
import json
import os
import re
import shlex
import sys
import threading
import time
from collections.abc import Callable
from datetime import UTC, date, datetime
from pathlib import Path
from urllib.parse import unquote, urlparse
import requests
from packaging.version import InvalidVersion, Version
from tradingagents.dataflows.files import replace_file
from tradingagents.default_config import _TRADINGAGENTS_HOME
RELEASES_URL = "https://api.github.com/repos/TauricResearch/TradingAgents/releases/latest"
NOTICES_URL = "https://api.tauric.ai/v1/announcements"
CACHE = Path(_TRADINGAGENTS_HOME) / "update_check.json"
CHECK_EVERY = 24 * 3600
TIMEOUT = 2.0
SUMMARY_LIMIT = 160
OPT_OUT = "TRADINGAGENTS_NO_UPDATE_CHECK"
# Control characters, C0 and C1, read as spaces; invisible spaces and direction marks are dropped.
CONTROL = re.compile(r"[\x00-\x1f\x7f-\x9f]")
INVISIBLE = re.compile(r"[\u061c\u200b\u200e\u200f\u202a-\u202e\u2066-\u2069\ufeff]")
DATE = re.compile(r"\d{4}-\d{2}-\d{2}")
def enabled() -> bool:
"""Whether a person is at the terminal and has not turned the check off."""
at_a_terminal = bool(sys.stdin and sys.stdin.isatty() and sys.stdout and sys.stdout.isatty())
return at_a_terminal and not os.environ.get("CI") and not os.environ.get(OPT_OUT)
# --- the cache and the check -------------------------------------------------------
def read_cache() -> dict:
"""The cache, keeping only fields of the shape this version writes."""
try:
data = json.loads(CACHE.read_text(encoding="utf-8"))
except (OSError, ValueError):
return {}
if not isinstance(data, dict):
return {}
shapes = {"checked_at": (int, float), "latest": dict, "notices": list}
return {key: data[key] for key, shape in shapes.items() if isinstance(data.get(key), shape)}
def due(cache: dict, now: float) -> bool:
checked = cache.get("checked_at")
return not isinstance(checked, (int, float)) or not 0 <= now - checked < CHECK_EVERY
def summary_of(body: str) -> str:
"""The release's one-line summary: its first paragraph on one line (sanitised when shown)."""
first = re.split(r"\n\s*\n", (body or "").strip(), maxsplit=1)[0]
text = " ".join(first.split())
if len(text) > SUMMARY_LIMIT:
text = text[: SUMMARY_LIMIT - 3].rsplit(" ", 1)[0] + "..."
return text
def parse_notices(payload) -> list | None:
"""The targeted notices in an announcements answer; None when it has no notices list."""
notices = payload.get("notices") if isinstance(payload, dict) else None
return notices if isinstance(notices, list) else None
def _get_json(url: str):
response = requests.get(url, timeout=TIMEOUT)
return response.json() if response.ok else None
def refresh(now: float | None = None) -> None:
"""Ask for the latest release and notices and record the answers. Never raises or prints;
a failed check keeps what was known and is not retried for a day."""
now = time.time() if now is None else now
cache = read_cache()
cache["checked_at"] = now
try:
release = _get_json(RELEASES_URL)
if isinstance(release, dict) and release.get("tag_name"):
cache["latest"] = {
"version": str(release["tag_name"]).removeprefix("v"),
"url": release.get("html_url"),
"summary": summary_of(release.get("body") or ""),
}
except Exception:
pass
try:
notices = parse_notices(_get_json(NOTICES_URL))
if notices is not None:
cache["notices"] = notices
except Exception:
pass
try:
CACHE.parent.mkdir(parents=True, exist_ok=True)
replace_file(CACHE, lambda temp: Path(temp).write_text(json.dumps(cache), encoding="utf-8"))
except Exception:
pass
def start_background_refresh() -> None:
"""Refresh the cache in a daemon thread when a check is due."""
if due(read_cache(), time.time()):
threading.Thread(target=refresh, name="tradingagents-update-check", daemon=True).start()
# --- what is shown ---------------------------------------------------------------
def _version(text) -> Version | None:
try:
return Version(str(text))
except (InvalidVersion, TypeError):
return None
def _plain(text) -> str:
return " ".join(CONTROL.sub(" ", INVISIBLE.sub("", str(text or ""))).split())
def _applies(notice: dict, running: Version, today: date) -> bool:
"""Whether a notice is for this version and unexpired; a bound or date that cannot be read
keeps it hidden. A notice is shown through its ``expires`` date (UTC)."""
low, high = notice.get("min_version"), notice.get("max_version")
if low is not None and ((floor := _version(low)) is None or running < floor):
return False
if high is not None and ((ceiling := _version(high)) is None or running > ceiling):
return False
expires = notice.get("expires")
if expires is None:
return True
try:
return isinstance(expires, str) and bool(DATE.fullmatch(expires)) and date.fromisoformat(expires) >= today
except ValueError:
return False
def messages(cache: dict, running: str, upgrade: Callable[[], str | None] = lambda: None) -> list[str]:
"""The lines to show: a newer release with its upgrade command (``upgrade()``, asked only
when there is one), then the notices for this version."""
current = _version(running)
if current is None:
return []
lines = []
latest = cache.get("latest") or {}
newest = _version(latest.get("version"))
if newest is not None and newest > current:
lines.append(f"TradingAgents {_plain(latest['version'])} is available (you have {_plain(running)}): "
f"{_plain(latest.get('summary'))}")
if latest.get("url"):
lines.append(_plain(latest["url"]))
command = upgrade()
if command:
lines.append(f"Upgrade: {_plain(command)}")
today = datetime.now(UTC).date()
for notice in cache.get("notices") or []:
if not (isinstance(notice, dict) and isinstance(notice.get("message"), str) and _plain(notice["message"])
and _applies(notice, current, today)):
continue
url = notice.get("url")
link = url if isinstance(url, str) and url.lower().startswith(("http://", "https://")) else None
lines.append(_plain(notice["message"]) + (f" {_plain(link)}" if link else ""))
return lines
def upgrade_command(dist=None) -> str | None:
"""The command that upgrades this install, from how pip or uv recorded it; None when unsure."""
if os.environ.get("TRADINGAGENTS_IN_DOCKER"):
return "git pull && docker compose build (on the host)"
if os.name == "nt": # the commands below are POSIX shell
return None
try:
dist = dist or importlib.metadata.distribution("tradingagents")
direct = json.loads(dist.read_text("direct_url.json") or "null")
installer = (dist.read_text("INSTALLER") or "").strip()
except (importlib.metadata.PackageNotFoundError, OSError, ValueError):
return None
if not isinstance(direct, dict) or not direct.get("url"):
return None
python = shlex.quote(sys.executable)
install = f"uv pip install --python {python}" if installer == "uv" else f"{python} -m pip install"
if "vcs_info" in direct:
return f"{install} -U git+{direct['url']}"
parsed = urlparse(direct["url"])
if parsed.scheme != "file":
return None
clone = Path(unquote(parsed.path))
if not (clone / ".git").exists():
return None
editable = (direct.get("dir_info") or {}).get("editable")
target = shlex.quote(str(clone))
return f"git -C {target} pull && {install} {'-e ' if editable else ''}{target}"
def show(console) -> None:
"""Print what's new from the cache, under the welcome panel."""
from rich.panel import Panel
from rich.text import Text
from tradingagents import __version__
lines = messages(read_cache(), running=__version__, upgrade=upgrade_command)
if lines:
console.print(Panel(Text("\n".join(lines)), border_style="cyan", padding=(1, 2), title="What's new"))
console.print()
+1
View File
@@ -16,6 +16,7 @@ dependencies = [
"langgraph>=1.2.12",
"langgraph-checkpoint-sqlite>=3.1.1",
"markdown-it-py>=4.0",
"packaging>=24.0",
"pandas>=3.0.6",
"python-dotenv>=1.0.0",
"pytz>=2025.2",
+3 -6
View File
@@ -54,8 +54,7 @@ class TestCliSkipsPromptsFromEnv(unittest.TestCase):
with mock.patch.dict(os.environ, env, clear=False), \
mock.patch.object(cli_selections, "DEFAULT_CONFIG", fake_cfg), \
mock.patch.object(cli_selections, "fetch_announcements", return_value=None), \
mock.patch.object(cli_selections, "display_announcements"), \
mock.patch.object(cli_selections.updates, "enabled", return_value=False), \
mock.patch.object(cli_selections, "get_ticker", return_value="AAPL"), \
mock.patch.object(cli_selections, "get_analysis_date", return_value="2026-05-29"), \
mock.patch.object(cli_selections, "select_analysts", return_value=[]), \
@@ -96,8 +95,7 @@ class TestResearchDepthSkippedFromEnv(unittest.TestCase):
with mock.patch.dict(os.environ, env, clear=False), \
mock.patch.object(cli_selections, "DEFAULT_CONFIG", fake_cfg), \
mock.patch.object(cli_selections, "fetch_announcements", return_value=None), \
mock.patch.object(cli_selections, "display_announcements"), \
mock.patch.object(cli_selections.updates, "enabled", return_value=False), \
mock.patch.object(cli_selections, "get_ticker", return_value="AAPL"), \
mock.patch.object(cli_selections, "get_analysis_date", return_value="2026-05-29"), \
mock.patch.object(cli_selections, "select_analysts", return_value=[]), \
@@ -125,8 +123,7 @@ class TestReasoningEffortSkippedFromEnv(unittest.TestCase):
with mock.patch.dict(os.environ, env, clear=False), \
mock.patch.object(cli_selections, "DEFAULT_CONFIG", fake_cfg), \
mock.patch.object(cli_selections, "fetch_announcements", return_value=None), \
mock.patch.object(cli_selections, "display_announcements"), \
mock.patch.object(cli_selections.updates, "enabled", return_value=False), \
mock.patch.object(cli_selections, "get_ticker", return_value="AAPL"), \
mock.patch.object(cli_selections, "get_analysis_date", return_value="2026-05-29"), \
mock.patch.object(cli_selections, "select_analysts", return_value=[]), \
+2 -17
View File
@@ -56,8 +56,7 @@ def test_flags_skip_exactly_their_own_steps(monkeypatch):
for step in ("get_ticker", "get_analysis_date", "select_analysts"):
monkeypatch.setattr(selections, step, no_prompt)
monkeypatch.setattr(selections, "fetch_announcements", lambda: None)
monkeypatch.setattr(selections, "display_announcements", lambda *a: None)
monkeypatch.setattr(selections.updates, "enabled", lambda: False)
chosen = selections._prompt_selections({}, FLAGS)
@@ -119,19 +118,6 @@ def test_save_and_show_answer_the_questions_after_the_run(monkeypatch, tmp_path)
@pytest.mark.unit
def test_an_announcement_does_not_wait_for_enter_without_a_terminal(monkeypatch):
from cli import announcements
from cli.display import console
def no_wait(*a):
raise AssertionError("waited for Enter with no terminal")
monkeypatch.setattr(announcements.getpass, "getpass", no_wait)
monkeypatch.setattr(announcements.sys.stdin, "isatty", lambda: False)
announcements.display_announcements(console, {"content": "Maintenance tonight", "require_attention": True})
@pytest.mark.unit
def test_a_missing_key_without_a_terminal_names_the_variable(monkeypatch, capsys):
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
@@ -200,8 +186,7 @@ def test_an_unknown_analyst_error_names_the_analysts_as_users_know_them():
def _selections_with_env(monkeypatch, keys=None):
for name, value in UNATTENDED_ENV.items():
monkeypatch.setenv(name, value)
monkeypatch.setattr(selections, "fetch_announcements", lambda: None)
monkeypatch.setattr(selections, "display_announcements", lambda *a: None)
monkeypatch.setattr(selections.updates, "enabled", lambda: False)
checked = keys if keys is not None else []
monkeypatch.setattr(selections, "ensure_api_key", lambda provider: checked.append(provider))
return checked
+2 -4
View File
@@ -95,8 +95,7 @@ def _answer_every_prompt(monkeypatch):
"""Drive the real selection flow, answering each prompt with a fixed value."""
import cli.main as m
monkeypatch.setattr(cli_selections, "fetch_announcements", lambda: [])
monkeypatch.setattr(cli_selections, "display_announcements", lambda *a: None)
monkeypatch.setattr(cli_selections.updates, "enabled", lambda: False)
monkeypatch.setattr(cli_selections, "get_ticker", lambda: "NVDA")
monkeypatch.setattr(cli_selections, "get_analysis_date", lambda: "2026-09-01")
monkeypatch.setattr(cli_selections, "ask_output_language", lambda default=None: "English")
@@ -148,8 +147,7 @@ def test_a_remembered_endpoint_is_offered_back(monkeypatch):
monkeypatch.setattr(cli_selections, "select_llm_provider", lambda default=None: ("openai_compatible", None))
monkeypatch.setattr(cli_selections, "prompt_openai_compatible_url",
lambda default=None: offered.setdefault("default", default) or "http://x/v1")
monkeypatch.setattr(cli_selections, "fetch_announcements", lambda: [])
monkeypatch.setattr(cli_selections, "display_announcements", lambda *a: None)
monkeypatch.setattr(cli_selections.updates, "enabled", lambda: False)
monkeypatch.setattr(cli_selections, "get_ticker", lambda: "NVDA")
monkeypatch.setattr(cli_selections, "get_analysis_date", lambda: "2026-09-01")
monkeypatch.setattr(cli_selections, "ask_output_language", lambda default=None: "English")
+337
View File
@@ -0,0 +1,337 @@
"""What's new reaches the person at a terminal: a newer release and Tauric's notices.
The check runs at most daily in a background thread that only writes a cache; the
notice shown is read from that cache, so a run never waits on the network, and
unattended runs contact no server at all.
"""
import io
import json
import os
from datetime import date
from types import SimpleNamespace
from unittest import mock
import pytest
from packaging.version import Version
from rich.console import Console
import cli.selections as cli_selections
import cli.updates as updates
class _Tty(io.StringIO):
def isatty(self):
return True
# --- who sees it -----------------------------------------------------------------
@pytest.mark.unit
@pytest.mark.parametrize("stdin_tty, stdout_tty, ci, opt_out, expected", [
(True, True, None, None, True),
(False, True, None, None, False),
(True, False, None, None, False),
(True, True, "true", None, False),
(True, True, None, "1", False),
])
def test_only_a_person_at_a_terminal_is_told(monkeypatch, stdin_tty, stdout_tty, ci, opt_out, expected):
monkeypatch.setattr("sys.stdin", _Tty() if stdin_tty else io.StringIO())
monkeypatch.setattr("sys.stdout", _Tty() if stdout_tty else io.StringIO())
for name, value in (("CI", ci), ("TRADINGAGENTS_NO_UPDATE_CHECK", opt_out)):
if value is None:
monkeypatch.delenv(name, raising=False)
else:
monkeypatch.setenv(name, value)
assert updates.enabled() is expected
ENV = {
"TRADINGAGENTS_LLM_PROVIDER": "openai",
"TRADINGAGENTS_DEEP_THINK_LLM": "gpt-6-sol",
"TRADINGAGENTS_QUICK_THINK_LLM": "gpt-6-luna",
"TRADINGAGENTS_OUTPUT_LANGUAGE": "English",
"TRADINGAGENTS_MAX_DEBATE_ROUNDS": "1",
"TRADINGAGENTS_MAX_RISK_ROUNDS": "1",
}
@pytest.mark.unit
def test_an_unattended_run_contacts_no_server(monkeypatch):
monkeypatch.setattr("sys.stdin", io.StringIO()) # no terminal
calls = []
monkeypatch.setattr(updates.requests, "get", lambda *a, **k: calls.append(a))
started = []
monkeypatch.setattr(updates.threading, "Thread", lambda *a, **k: started.append(1))
with mock.patch.dict(os.environ, ENV, clear=False), \
mock.patch.object(cli_selections, "ensure_api_key"):
cli_selections.get_user_selections({"ticker": "AAPL", "date": "2026-09-01", "analysts": "market"})
assert calls == [] and started == []
# --- what it says -----------------------------------------------------------------
LATEST = {"version": "0.7.0", "url": "https://github.com/TauricResearch/TradingAgents/releases/tag/v0.7.0",
"summary": "Update notices and a ChatGPT plan provider."}
@pytest.mark.unit
@pytest.mark.parametrize("running, shown", [
("0.6.0", True), ("0.7.0.dev0", True), # a development build of the release is older than it
("0.7.0", False), ("0.8.0.dev0", False), ("not-a-version", False),
])
def test_a_newer_release_is_announced_once_in_the_notice(running, shown):
lines = updates.messages({"latest": LATEST}, running=running, upgrade=lambda: None)
assert any("0.7.0 is available" in line for line in lines) is shown
if shown:
text = "\n".join(lines)
assert f"you have {running}" in text and LATEST["url"] in text and LATEST["summary"] in text
@pytest.mark.unit
def test_the_notice_names_the_upgrade_command():
lines = updates.messages({"latest": LATEST}, running="0.6.0", upgrade=lambda: "git -C /src pull && pip install /src")
assert "Upgrade: git -C /src pull && pip install /src" in "\n".join(lines)
@pytest.mark.unit
def test_remote_text_cannot_style_or_break_the_terminal():
body = "[bold red]Grab it[/bold red] now\x1b[2J with `--html`, a very long line " + "x" * 400 + "\n\nSecond paragraph."
summary = updates.summary_of(body)
assert "\n" not in summary and "Second paragraph" not in summary and len(summary) <= 160
lines = updates.messages({"latest": {**LATEST, "summary": summary}}, running="0.6.0", upgrade=lambda: None)
shown = "\n".join(lines)
assert "[bold red]Grab it" in shown and "\x1b" not in shown # shown as plain text, controls stripped
@pytest.mark.unit
def test_notices_reach_only_the_versions_they_name():
cache = {"notices": [
{"id": "a", "message": "For everyone", "url": None},
{"id": "b", "message": "Old installs: news is fixed", "max_version": "0.6.0"},
{"id": "c", "message": "New installs only", "min_version": "0.7.0"},
{"id": "d", "message": "Expired", "expires": "2020-01-01"},
{"id": "e", "message": "[link=x]markup[/link]"},
"a bare string is not a notice",
]}
text = "\n".join(updates.messages(cache, running="0.6.0", upgrade=lambda: None))
assert "For everyone" in text and "Old installs" in text
assert "New installs only" not in text and "Expired" not in text
assert "[link=x]markup[/link]" in text and "bare string" not in text
@pytest.mark.unit
def test_a_notice_shows_through_its_bounds_and_date_and_hides_when_they_cannot_be_read():
today = date(2026, 10, 16)
applies = lambda **notice: updates._applies(notice, Version("0.7.0"), today) # noqa: E731
assert applies(min_version="0.7.0", max_version="0.7.0", expires="2026-10-16")
assert applies(expires=None) and applies(min_version=None, max_version=None)
assert not applies(expires="2026-10-15") and not applies(min_version="0.7.1") and not applies(max_version="0.6.9")
for unreadable in ("", "seven", [], False):
assert not applies(min_version=unreadable) and not applies(max_version=unreadable)
for unreadable in ("", "2026-10-16junk", "20261020", "2026-W42-5", "2026-10-16T23:00-08:00", 20261016):
assert not applies(expires=unreadable)
assert not updates._applies({"min_version": "0.7.0"}, Version("0.7.0.dev0"), today)
@pytest.mark.unit
def test_a_notice_is_shown_as_plain_text_with_only_a_web_link():
cache = {"notices": [
{"id": "a", "message": ":warning: [b]x[/b] \u202eev\u200bil", "url": "https://tauric.ai/a"},
{"id": "b", "message": "Local link", "url": "file:///etc/passwd"},
{"id": "e", "message": "Malformed link", "url": "https://[docs"},
{"id": "c", "message": {"not": "text"}},
{"id": "d", "message": "\x1b\x07"},
]}
assert updates.messages(cache, running="0.7.0") == [":warning: [b]x[/b] evil https://tauric.ai/a", "Local link",
"Malformed link https://[docs"]
console = Console(record=True, width=120, color_system=None)
with mock.patch.object(updates, "read_cache", lambda: cache), mock.patch("tradingagents.__version__", "0.7.0"):
updates.show(console)
assert ":warning: [b]x[/b] evil" in console.export_text()
@pytest.mark.unit
def test_notices_are_read_from_their_own_key():
# Notices have their own key, so ``announcements`` stays a list of strings.
payload = {"announcements": ["Plain text for older clients"],
"notices": [{"id": "x", "message": "Targeted", "max_version": "0.6.0"}]}
assert updates.parse_notices(payload) == payload["notices"]
assert updates.parse_notices({"announcements": ["old shape only"]}) is None
assert updates.parse_notices({"notices": []}) == []
# --- how it is checked ---------------------------------------------------------------
@pytest.mark.unit
def test_a_check_is_due_once_a_day():
now = 1_000_000.0
assert updates.due({}, now)
assert not updates.due({"checked_at": now - 3600}, now)
assert updates.due({"checked_at": now - 25 * 3600}, now)
def _response(payload, status=200):
return SimpleNamespace(status_code=status, ok=status < 400, json=lambda: payload)
@pytest.mark.unit
def test_a_check_records_the_latest_release_and_notices(tmp_path, monkeypatch):
cache_file = tmp_path / "update_check.json"
monkeypatch.setattr(updates, "CACHE", cache_file)
def get(url, timeout):
assert timeout <= 3
if "api.github.com" in url:
return _response({"tag_name": "v0.7.0", "html_url": LATEST["url"],
"body": "Update notices and a ChatGPT plan provider.\n\n## Added"})
return _response({"announcements": ["old"], "notices": [{"id": "n", "message": "hi"}]})
monkeypatch.setattr(updates.requests, "get", get)
updates.refresh(now=5000.0)
cache = json.loads(cache_file.read_text())
assert cache["checked_at"] == 5000.0
assert cache["latest"] == {"version": "0.7.0", "url": LATEST["url"], "summary": LATEST["summary"]}
assert cache["notices"] == [{"id": "n", "message": "hi"}]
@pytest.mark.unit
def test_a_failed_check_keeps_what_was_known_and_waits_a_day(tmp_path, monkeypatch, capsys):
cache_file = tmp_path / "update_check.json"
cache_file.write_text(json.dumps({"checked_at": 1.0, "latest": LATEST, "notices": []}))
monkeypatch.setattr(updates, "CACHE", cache_file)
def get(url, timeout):
raise updates.requests.ConnectionError("offline")
monkeypatch.setattr(updates.requests, "get", get)
updates.refresh(now=9000.0) # never raises
cache = json.loads(cache_file.read_text())
assert cache["checked_at"] == 9000.0 and cache["latest"] == LATEST
assert capsys.readouterr() == ("", "") # the check never prints
@pytest.mark.unit
def test_only_a_notices_list_replaces_the_known_notices(tmp_path, monkeypatch):
cache_file = tmp_path / "update_check.json"
known = [{"id": "n", "message": "hi"}]
cache_file.write_text(json.dumps({"checked_at": 1.0, "notices": known}))
monkeypatch.setattr(updates, "CACHE", cache_file)
for answer in (_response({"detail": "Not Found"}, 404), _response({"error": "portal"}), _response(["x"])):
monkeypatch.setattr(updates.requests, "get", lambda url, timeout, answer=answer: answer)
updates.refresh(now=9000.0)
assert json.loads(cache_file.read_text())["notices"] == known
monkeypatch.setattr(updates.requests, "get", lambda url, timeout: _response({"announcements": [], "notices": []}))
updates.refresh(now=99000.0)
assert json.loads(cache_file.read_text())["notices"] == []
@pytest.mark.unit
def test_an_unreadable_cache_is_no_news(tmp_path, monkeypatch):
cache_file = tmp_path / "update_check.json"
cache_file.write_text("{not json")
monkeypatch.setattr(updates, "CACHE", cache_file)
assert updates.read_cache() == {}
# --- the upgrade command for how it was installed ----------------------------------
def _dist(direct=None, installer="pip"):
files = {"direct_url.json": json.dumps(direct) if direct else None, "INSTALLER": installer}
return SimpleNamespace(read_text=lambda name: files.get(name))
@pytest.mark.unit
def test_the_upgrade_command_follows_the_install(tmp_path, monkeypatch):
clone = tmp_path / "TradingAgents"
(clone / ".git").mkdir(parents=True)
monkeypatch.delenv("TRADINGAGENTS_IN_DOCKER", raising=False)
monkeypatch.setattr(updates.sys, "executable", "/venv/bin/python")
url = clone.as_uri()
editable = updates.upgrade_command(_dist({"url": url, "dir_info": {"editable": True}}))
assert editable == f"git -C {clone} pull && /venv/bin/python -m pip install -e {clone}"
local = updates.upgrade_command(_dist({"url": url, "dir_info": {}}, installer="uv"))
assert local == f"git -C {clone} pull && uv pip install --python /venv/bin/python {clone}"
vcs = updates.upgrade_command(_dist({"url": "https://github.com/TauricResearch/TradingAgents.git",
"vcs_info": {"vcs": "git", "commit_id": "abc"}}))
assert vcs == "/venv/bin/python -m pip install -U git+https://github.com/TauricResearch/TradingAgents.git"
@pytest.mark.unit
def test_no_command_is_guessed_without_a_clone(tmp_path, monkeypatch):
monkeypatch.delenv("TRADINGAGENTS_IN_DOCKER", raising=False)
gone = (tmp_path / "deleted-clone").as_uri()
assert updates.upgrade_command(_dist({"url": gone, "dir_info": {}})) is None
assert updates.upgrade_command(_dist(None)) is None
monkeypatch.setenv("TRADINGAGENTS_IN_DOCKER", "1")
assert updates.upgrade_command(_dist(None)) == "git pull && docker compose build (on the host)"
@pytest.mark.unit
def test_whats_new_never_waits_for_the_person(tmp_path, monkeypatch):
cache_file = tmp_path / "update_check.json"
cache_file.write_text(json.dumps({"latest": LATEST, "notices": [{"id": "x", "message": "Notice"}]}))
monkeypatch.setattr(updates, "CACHE", cache_file)
monkeypatch.setattr("builtins.input", lambda *a: (_ for _ in ()).throw(AssertionError("asked for input")))
from rich.console import Console
out = Console(file=io.StringIO(), width=200)
updates.show(out)
assert "Notice" in out.file.getvalue()
@pytest.mark.unit
@pytest.mark.parametrize("stored", [
{"latest": "x"}, {"latest": ["0.9.0"]}, {"notices": 5}, {"notices": {"id": "x"}}, {"checked_at": "today"}, [],
])
def test_a_cache_of_the_wrong_shape_is_no_news(tmp_path, monkeypatch, stored):
cache_file = tmp_path / "update_check.json"
cache_file.write_text(json.dumps(stored))
monkeypatch.setattr(updates, "CACHE", cache_file)
assert updates.messages(updates.read_cache(), running="0.6.0", upgrade=lambda: None) == []
assert updates.due(updates.read_cache(), 1_000_000.0)
@pytest.mark.unit
def test_a_check_dated_in_the_future_is_due():
assert updates.due({"checked_at": 2_000_000.0}, 1_000_000.0)
@pytest.mark.unit
def test_a_notice_with_an_unreadable_expiry_is_not_shown():
cache = {"notices": [{"message": "Bad date", "expires": "next week"},
{"message": "Number", "expires": 20991231},
{"message": "Timestamp", "expires": "2999-01-01T00:00:00Z"},
{"message": "A date", "expires": "2999-01-01"}]}
text = "\n".join(updates.messages(cache, running="0.6.0", upgrade=lambda: None))
assert "Bad date" not in text and "Number" not in text and "Timestamp" not in text and "A date" in text
@pytest.mark.unit
def test_remote_text_cannot_carry_c1_or_direction_controls():
lines = updates.messages({"notices": [{"message": "A\x9b2Jb‮evil"}]}, running="0.6.0", upgrade=lambda: None)
assert lines == ["A 2Jbevil"] # the C1 control reads as a space, the override is dropped
@pytest.mark.unit
def test_a_uv_git_install_is_upgraded_with_uv(monkeypatch):
monkeypatch.delenv("TRADINGAGENTS_IN_DOCKER", raising=False)
monkeypatch.setattr(updates.sys, "executable", "/venv/bin/python")
vcs = updates.upgrade_command(_dist({"url": "https://github.com/TauricResearch/TradingAgents.git",
"vcs_info": {"vcs": "git"}}, installer="uv"))
assert vcs == "uv pip install --python /venv/bin/python -U git+https://github.com/TauricResearch/TradingAgents.git"
@pytest.mark.unit
def test_no_command_is_offered_on_windows(monkeypatch):
monkeypatch.delenv("TRADINGAGENTS_IN_DOCKER", raising=False)
monkeypatch.setattr(updates.os, "name", "nt")
assert updates.upgrade_command(_dist({"url": "https://github.com/TauricResearch/TradingAgents.git",
"vcs_info": {"vcs": "git"}})) is None
@pytest.mark.unit
def test_the_upgrade_command_is_worked_out_only_for_a_newer_release():
asked = []
updates.messages({"latest": LATEST}, running=LATEST["version"], upgrade=lambda: asked.append(1))
assert asked == []