fix: improve show handling, cache validation and tests

Add extensive tests for the `show` command (happy path, empty/expired/malformed cache, out-of-range, zero/negative indices) and a test helper to write cache fixtures. Harden twitter_cli.cache._load_cache to validate payload types (dict/tweets list), filter non-dict entries, and treat malformed payloads as empty; also handle missing tweet id when resolving an index. Refactor CLI output logic by extracting _emit_tweet_detail and reuse it for both `tweet` and `show`; enforce 1-based indices for `show` via click.IntRange(1) and expand the "no cached results" error text to mention other list commands. These changes improve robustness against corrupted caches and increase test coverage for cache-based behavior.
This commit is contained in:
Pleasurecruise
2026-03-12 22:26:52 +00:00
parent 5335516d57
commit 72f62cedea
3 changed files with 123 additions and 17 deletions

View File

@@ -1,5 +1,8 @@
from __future__ import annotations
import json
import time
from click.testing import CliRunner
import pytest
from rich.console import Console
@@ -538,3 +541,105 @@ def test_cli_compact_mode(tmp_path, tweet_factory) -> None:
assert '"@alice"' in result.output
# Compact output should NOT have full metrics keys
assert '"metrics"' not in result.output
def _write_cache(cache_file, tweets, created_at=None):
"""Write a test cache file."""
if created_at is None:
created_at = time.time()
entries = [
{"index": i + 1, "id": t.id, "author": t.author.screen_name, "text": t.text[:80]}
for i, t in enumerate(tweets)
]
payload = {"created_at": created_at, "tweets": entries}
cache_file.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
def test_show_happy_path(monkeypatch, tmp_path, tweet_factory):
"""show <N> resolves cached index and fetches tweet detail."""
tw = tweet_factory("42", text="hello world")
cache_file = tmp_path / "last_results.json"
_write_cache(cache_file, [tweet_factory("10"), tw]) # tw is index 2
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
class FakeClient:
def fetch_tweet_detail(self, tweet_id, count):
assert tweet_id == "42"
return [tw]
monkeypatch.setattr("twitter_cli.cli._get_client", lambda config=None, quiet=False: FakeClient())
monkeypatch.setattr("twitter_cli.cli.load_config", lambda: {})
runner = CliRunner()
result = runner.invoke(cli, ["show", "2"])
assert result.exit_code == 0
def test_show_empty_cache(monkeypatch, tmp_path):
"""show fails with a helpful message when no cache exists."""
cache_file = tmp_path / "last_results.json"
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
runner = CliRunner()
result = runner.invoke(cli, ["show", "1"])
assert result.exit_code != 0
assert "No cached results" in result.output
def test_show_out_of_range(monkeypatch, tmp_path, tweet_factory):
"""show fails with out-of-range message when index exceeds cache size."""
cache_file = tmp_path / "last_results.json"
_write_cache(cache_file, [tweet_factory("1")])
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
runner = CliRunner()
result = runner.invoke(cli, ["show", "99"])
assert result.exit_code != 0
assert "out of range" in result.output
assert "1" in result.output # cache has 1 tweet
def test_show_expired_cache(monkeypatch, tmp_path, tweet_factory):
"""show treats an expired cache the same as no cache."""
cache_file = tmp_path / "last_results.json"
expired_time = time.time() - 7200 # 2 hours ago
_write_cache(cache_file, [tweet_factory("1")], created_at=expired_time)
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
runner = CliRunner()
result = runner.invoke(cli, ["show", "1"])
assert result.exit_code != 0
assert "No cached results" in result.output
def test_show_rejects_zero_index(monkeypatch, tmp_path):
"""show rejects index=0 because indices are 1-based."""
cache_file = tmp_path / "last_results.json"
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
runner = CliRunner()
result = runner.invoke(cli, ["show", "0"])
assert result.exit_code != 0
def test_show_rejects_negative_index(monkeypatch, tmp_path):
"""show rejects negative indices."""
cache_file = tmp_path / "last_results.json"
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
runner = CliRunner()
result = runner.invoke(cli, ["show", "-1"])
assert result.exit_code != 0
def test_show_malformed_cache_treated_as_empty(monkeypatch, tmp_path):
"""show handles a corrupted cache file gracefully."""
cache_file = tmp_path / "last_results.json"
cache_file.write_text("not valid json{{}", encoding="utf-8")
monkeypatch.setattr("twitter_cli.cache._CACHE_FILE", cache_file)
runner = CliRunner()
result = runner.invoke(cli, ["show", "1"])
assert result.exit_code != 0
assert "No cached results" in result.output

View File

@@ -38,9 +38,14 @@ def _load_cache() -> Optional[List[dict]]:
if not _CACHE_FILE.exists():
return None
payload = json.loads(_CACHE_FILE.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
return None
if time.time() - payload.get("created_at", 0) > _TTL:
return None
return payload.get("tweets", [])
entries = payload.get("tweets", [])
if not isinstance(entries, list):
return None
return [e for e in entries if isinstance(e, dict)]
except (OSError, json.JSONDecodeError):
return None
@@ -52,7 +57,8 @@ def get_tweet_id_by_index(index: int) -> Optional[str]:
return None
for entry in entries:
if entry.get("index") == index:
return str(entry["id"])
tweet_id = entry.get("id")
return str(tweet_id) if tweet_id is not None else None
return None

View File

@@ -704,6 +704,12 @@ def tweet(ctx, tweet_id, max_count, full_text, as_json, as_yaml):
except RuntimeError as exc:
_exit_with_error(exc)
_emit_tweet_detail(tweets, compact=compact, as_json=as_json, as_yaml=as_yaml, full_text=full_text)
def _emit_tweet_detail(tweets, compact, as_json, as_yaml, full_text):
# type: (list, bool, bool, bool, bool) -> None
"""Render tweet detail + replies in the requested output format."""
if compact:
click.echo(tweets_to_compact_json(tweets))
return
@@ -720,7 +726,7 @@ def tweet(ctx, tweet_id, max_count, full_text, as_json, as_yaml):
@cli.command()
@click.argument("index", type=int)
@click.argument("index", type=click.IntRange(1))
@click.option("--max", "-n", "max_count", type=int, default=None, help="Max replies to fetch.")
@click.option("--full-text", is_flag=True, help="Show full reply text in table output.")
@structured_output_options
@@ -735,7 +741,8 @@ def show(ctx, index, max_count, full_text, as_json, as_yaml):
cache_size = get_cache_size()
if cache_size == 0:
raise click.UsageError(
"No cached results found. Run `twitter feed` or `twitter search` first."
"No cached results found. Run `twitter feed`, `twitter search`, "
"`twitter bookmarks`, or another list command first."
)
raise click.UsageError(
"Index %d is out of range (cache has %d tweets)." % (index, cache_size)
@@ -755,19 +762,7 @@ def show(ctx, index, max_count, full_text, as_json, as_yaml):
except RuntimeError as exc:
_exit_with_error(exc)
if compact:
click.echo(tweets_to_compact_json(tweets))
return
if emit_structured(tweets_to_data(tweets), as_json=as_json, as_yaml=as_yaml):
return
if tweets:
print_tweet_detail(tweets[0], console)
if len(tweets) > 1:
console.print("\n💬 Replies:")
print_tweet_table(tweets[1:], console, title="💬 Replies — %d" % (len(tweets) - 1), full_text=full_text)
console.print()
_emit_tweet_detail(tweets, compact=compact, as_json=as_json, as_yaml=as_yaml, full_text=full_text)
@cli.command()