nielsr's picture
nielsr HF Staff
Prototype PwC MCP Space from pwc-cli@6a547659
ee0b555 verified
Raw History Blame Contribute Delete
106 kB
from __future__ import annotations
import argparse
import importlib.util
import io
import json
import sys
import urllib.error
import urllib.request
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from pwc_cli.cli import UsageError, build_parser, main # noqa: E402
from pwc_cli.skills import build_skill_md # noqa: E402
from pwc_cli.transport import ( # noqa: E402
Client,
HTTPStatusError,
Response,
ResponseError,
)
INSTALLER_SPEC = importlib.util.spec_from_file_location(
"pwc_cli_installer", ROOT / "install.py"
)
assert INSTALLER_SPEC and INSTALLER_SPEC.loader
installer = importlib.util.module_from_spec(INSTALLER_SPEC)
INSTALLER_SPEC.loader.exec_module(installer)
def test_complete_parser_omits_administrative_commands():
help_text = build_parser().format_help()
assert (
"{search,paper,task,method,conference,organization,framework,benchmark,skills,version,auth}"
in help_text
)
for command in ("add-external", "cron", "embedding", "github-issue"):
assert command not in help_text
assert build_parser().parse_args(["paper", "lineage", "list", "2501.1"]).paper
assert (
build_parser()
.parse_args(["paper", "info", "2501.1", "--include-evals"])
.include_evals
)
paper_filters = build_parser().parse_args(
[
"paper",
"list",
"--task",
"Image Classification",
"--method",
"Vision Transformer",
"--conference",
"CVPR 2026",
"--framework",
"vLLM",
"--organization",
"Qwen",
]
)
assert paper_filters.task == "Image Classification"
assert paper_filters.method == "Vision Transformer"
assert paper_filters.conference == "CVPR 2026"
assert paper_filters.framework == "vLLM"
assert paper_filters.organization == "Qwen"
assert (
build_parser().parse_args(["task", "--name", "scene-text-recognition"]).name
== "scene-text-recognition"
)
assert build_parser().parse_args(["task", "list", "--area", "Vision"]).area
assert build_parser().parse_args(["task", "list", "--group-by-area"]).group_by_area
assert build_parser().parse_args(["method", "list", "--area", "Audio"]).area
assert build_parser().parse_args(["method", "--name", "Transformer"]).name
assert build_parser().parse_args(["conference", "list", "--year", "2025"]).year
assert build_parser().parse_args(["conference", "--name", "CVPR 2025"]).name
assert build_parser().parse_args(["organization", "--name", "NVIDIA"]).name
assert (
build_parser()
.parse_args(["organization", "list", "--featured-only"])
.featured_only
)
assert build_parser().parse_args(["framework", "--name", "vLLM"]).name
assert (
build_parser().parse_args(["framework", "list", "--platform", "gpu"]).platform
== "gpu"
)
def test_generated_skill_matches_installed_cli_version_and_commands():
skill = build_skill_md()
assert "name: pwc-cli" in skill
assert "Generated with `pwc v0.4.6`" in skill
assert "`pwc search QUERY" in skill
assert "--include-evals" in skill
assert "[--organization ORGANIZATION]" in skill
assert "[--author AUTHOR]" in skill
assert "When the user identifies an author" in skill
assert "`pwc organization [--name NAME] [--json]`" in skill
assert "`pwc framework list [--domain DOMAIN]" in skill
assert "`pwc benchmark --name NAME" in skill
assert "[--max-parameters SIZE]" in skill
assert "`pwc skills add" in skill
assert "Publication date ranges are inclusive" in skill
assert "must not be later than `--end-date`" in skill
def test_published_skill_documents_x_thread_edits():
skill = (ROOT / "SKILL.md").read_text(encoding="utf-8")
assert "x_threads/x_threads" in skill
assert "authors or their organization" in skill
def test_published_skill_advertises_inclusive_date_ranges():
skill = (ROOT / "SKILL.md").read_text(encoding="utf-8")
assert "--start-date YYYY-MM-DD --end-date YYYY-MM-DD" in skill
assert "--published-after" not in skill
assert "--time today|week|month|all_time" not in skill
assert "--max-parameters SIZE" in skill
assert "inclusive parameter" in skill
def test_skills_add_installs_project_skill(monkeypatch, tmp_path, capsys):
monkeypatch.chdir(tmp_path)
assert main(["skills", "add"]) == 0
skill_file = tmp_path / ".agents" / "skills" / "pwc-cli" / "SKILL.md"
assert skill_file.read_text() == build_skill_md()
assert str(skill_file.parent) in capsys.readouterr().out
def test_skills_add_global_and_claude_uses_user_directories(
monkeypatch, tmp_path, capsys
):
monkeypatch.setenv("HOME", str(tmp_path))
assert main(["skills", "add", "--global", "--claude"]) == 0
central = tmp_path / ".agents" / "skills" / "pwc-cli"
claude = tmp_path / ".claude" / "skills" / "pwc-cli"
assert (central / "SKILL.md").is_file()
assert claude.is_symlink()
assert claude.resolve() == central
output = capsys.readouterr().out
assert str(central) in output
assert str(claude) in output
def test_skills_add_protects_existing_install_and_force_replaces_it(
monkeypatch, tmp_path, capsys
):
monkeypatch.chdir(tmp_path)
existing = tmp_path / ".agents" / "skills" / "pwc-cli"
existing.mkdir(parents=True)
(existing / "old.txt").write_text("keep unless forced")
assert main(["skills", "add"]) == 2
assert (existing / "old.txt").is_file()
assert "rerun with --force" in capsys.readouterr().err
assert main(["skills", "add", "--force"]) == 0
assert not (existing / "old.txt").exists()
assert (existing / "SKILL.md").is_file()
def test_skills_add_supports_custom_harness_directory_and_rejects_mixed_scope(
tmp_path, capsys
):
custom = tmp_path / "agent-skills"
assert main(["skills", "add", "--dest", str(custom)]) == 0
assert (custom / "pwc-cli" / "SKILL.md").is_file()
assert main(["skills", "add", "--dest", str(custom), "--global"]) == 2
assert "--dest cannot be combined" in capsys.readouterr().err
def test_installer_downloads_from_this_repository_release_origin():
source = (ROOT / "install.py").read_text()
assert "github.com/huggingface/pwc-cli/releases/download" in source
assert "github.com/paperswithcode/paperswithcode.co" not in source
def test_installer_version_pin_is_optional_and_explicit_pin_skips_lookup(monkeypatch):
assert installer.parse_args([]).version is None
monkeypatch.setattr(
installer, "_open", lambda _url: (_ for _ in ()).throw(AssertionError())
)
assert installer.resolve_version("1.2.3", "unused") == "1.2.3"
def test_installer_resolves_valid_latest_release_tag(monkeypatch):
class Response(io.BytesIO):
def __enter__(self):
return self
def __exit__(self, *_args):
self.close()
monkeypatch.setattr(
installer,
"_open",
lambda _url: Response(b'{"tag_name": "pwc-cli-1.2.3"}'),
)
assert installer.resolve_version(None, "https://example.test/latest") == "1.2.3"
def test_installer_release_lookup_network_error_is_one_line_exit_3(monkeypatch, capsys):
monkeypatch.setattr(
installer.urllib.request,
"urlopen",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
urllib.error.URLError("offline\nnow")
),
)
assert installer.main([]) == 3
assert capsys.readouterr().err == "pwc installer: network error: offline now\n"
def test_installer_artifact_and_checksum_network_errors_exit_3(
monkeypatch, tmp_path, capsys
):
monkeypatch.setattr(installer.platform, "system", lambda: "Linux")
monkeypatch.setattr(installer.platform, "machine", lambda: "x86_64")
monkeypatch.setattr(
installer,
"_download",
lambda *_args: (_ for _ in ()).throw(installer.NetworkError("artifact")),
)
assert installer.main(["--version", "1.2.3", "--prefix", str(tmp_path)]) == 3
assert capsys.readouterr().err == "pwc installer: network error: artifact\n"
monkeypatch.setattr(
installer, "_download", lambda _url, destination: destination.write_bytes(b"x")
)
monkeypatch.setattr(
installer,
"_read_checksum",
lambda *_args: (_ for _ in ()).throw(installer.NetworkError("checksum")),
)
assert installer.main(["--version", "1.2.3", "--prefix", str(tmp_path)]) == 3
assert capsys.readouterr().err == "pwc installer: network error: checksum\n"
def test_installer_reports_path_and_persistent_instruction(tmp_path, capsys):
bin_directory = tmp_path / ".local" / "bin"
installer.report_path(bin_directory, f"/usr/bin:{bin_directory}")
captured = capsys.readouterr()
assert captured.out == f"{bin_directory} is on PATH\n"
assert captured.err == ""
installer.report_path(bin_directory, "/usr/bin")
captured = capsys.readouterr()
assert captured.out == ""
assert captured.err == (
f"warning: {bin_directory} is not on PATH\n"
f'Add this to ~/.profile: export PATH={bin_directory}:"$PATH"\n'
)
def test_installer_default_path_instruction_uses_expandable_home(capsys):
installer.report_path(Path.home() / ".local" / "bin", "/usr/bin")
assert capsys.readouterr().err == (
"warning: ~/.local/bin is not on PATH\n"
'Add this to ~/.profile: export PATH="$HOME/.local/bin:$PATH"\n'
)
def test_installer_replaces_binary_atomically(tmp_path):
artifact = tmp_path / "artifact"
artifact.write_bytes(b"new executable")
destination = tmp_path / "bin" / "pwc"
destination.parent.mkdir()
destination.write_bytes(b"old executable")
installer._install_atomically(artifact, destination)
assert destination.read_bytes() == b"new executable"
assert destination.stat().st_mode & 0o111
assert list(destination.parent.iterdir()) == [destination]
def test_benchmark_limit_alias_preserves_page_size_destination():
parser = build_parser()
assert parser.parse_args(["benchmark", "list", "--limit", "7"]).page_size == 7
assert parser.parse_args(["benchmark", "list", "--page-size", "8"]).page_size == 8
def test_paper_list_forwards_composable_catalog_filters(monkeypatch):
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
return Response(
b'{"count":0,"results":[],"applied_filters":'
b'{"task":"Image Classification","method":"Vision Transformer",'
b'"conference":"CVPR 2026","framework":"vLLM",'
b'"organization":"Qwen","authors":['
b'{"id":"264","name":"Kaiming He","hf_username":null},'
b'{"id":"26405","name":"Yilun Du","hf_username":"yilundu"}]}}',
{},
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert (
main(
[
"paper",
"list",
"--task",
"Image Classification",
"--method",
"Vision Transformer",
"--conference",
"CVPR 2026",
"--framework",
"vLLM",
"--organization",
"Qwen",
"--author",
"Kaiming He",
"--author",
"@yilundu",
]
)
== 0
)
assert calls == [
(
"papers/",
{
"page": 1,
"page_size": 20,
"search": None,
"start_date": None,
"end_date": None,
"task": "Image Classification",
"method": "Vision Transformer",
"conference": "CVPR 2026",
"framework": "vLLM",
"organization": "Qwen",
"author": ["Kaiming He", "@yilundu"],
"latest_only": True,
"order_by": "trending",
"order_dir": "desc",
"include_resources": False,
},
)
]
def test_paper_list_rejects_more_than_ten_authors_before_request(monkeypatch, capsys):
class Client:
def __init__(self):
pass
def get(self, path, params=None):
raise AssertionError("author limit must be checked before the request")
monkeypatch.setattr("pwc_cli.cli.Client", Client)
arguments = ["paper", "list"]
for author_id in range(1, 12):
arguments.extend(("--author", str(author_id)))
assert main(arguments) == 2
assert "at most 10 author filters" in capsys.readouterr().err.lower()
def test_search_and_paper_list_forward_inclusive_date_ranges(monkeypatch):
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
return Response(
b'{"count":0,"results":[],"applied_filters":'
b'{"start_date":"2026-08-01","end_date":"2026-08-21"}}',
{},
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert (
main(
[
"search",
"date filtering",
"--start-date",
"2026-08-01",
"--end-date",
"2026-08-21",
]
)
== 0
)
assert (
main(
[
"paper",
"list",
"--start-date",
"2026-08-01",
"--end-date",
"2026-08-21",
]
)
== 0
)
assert calls[0][1]["start_date"].isoformat() == "2026-08-01"
assert calls[0][1]["end_date"].isoformat() == "2026-08-21"
assert calls[1][1]["start_date"].isoformat() == "2026-08-01"
assert calls[1][1]["end_date"].isoformat() == "2026-08-21"
def test_date_range_rejects_an_end_before_its_start(capsys):
assert (
main(
[
"search",
"date filtering",
"--start-date",
"2026-08-21",
"--end-date",
"2026-08-01",
]
)
== 2
)
assert "start date must be on or before end date" in capsys.readouterr().err
def test_paper_list_rejects_server_that_does_not_confirm_filters(monkeypatch, capsys):
class Client:
def __init__(self):
pass
def get(self, path, params=None):
return Response(b'{"count":20,"results":[]}', {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["paper", "list", "--organization", "Qwen"]) == 4
assert "server must be upgraded" in capsys.readouterr().err
def test_paper_list_rejects_server_that_does_not_confirm_authors(monkeypatch, capsys):
class Client:
def __init__(self):
pass
def get(self, path, params=None):
return Response(b'{"count":20,"results":[],"applied_filters":{}}', {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["paper", "list", "--author", "Kaiming He"]) == 4
assert "did not confirm" in capsys.readouterr().err
def test_paper_list_rejects_partial_author_confirmation(monkeypatch, capsys):
class Client:
def __init__(self):
pass
def get(self, path, params=None):
return Response(
b'{"count":0,"results":[],"applied_filters":{"authors":['
b'{"id":"264","name":"Kaiming He","hf_username":null}]}}',
{},
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert (
main(
[
"paper",
"list",
"--author",
"Kaiming He",
"--author",
"@yilundu",
]
)
== 4
)
assert "did not confirm" in capsys.readouterr().err
def test_paper_list_accepts_canonical_confirmation_for_zero_padded_id(
monkeypatch, capsys
):
class Client:
def __init__(self):
pass
def get(self, path, params=None):
return Response(
b'{"count":0,"results":[],"applied_filters":{"authors":['
b'{"id":"264","name":"Kaiming He","hf_username":null}]}}',
{},
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["paper", "list", "--author", "000264"]) == 0
assert capsys.readouterr().err == ""
def test_paper_list_reports_ambiguous_author_as_usage_error(monkeypatch, capsys):
class Client:
def __init__(self):
pass
def get(self, path, params=None):
raise HTTPStatusError(
409,
json.dumps(
{
"detail": {
"code": "ambiguous_author",
"reference": "Alex Kim",
"candidates": [
{
"id": "41",
"name": "Alex Kim",
"hf_username": "alex-one",
},
{
"id": "72",
"name": "Alex Kim",
"hf_username": None,
},
],
}
}
),
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["paper", "list", "--author", "Alex Kim"]) == 2
error = capsys.readouterr().err
assert "Alex Kim is ambiguous" in error
assert "--author 41" in error
assert "@alex-one" in error
assert "--author 72" in error
def test_benchmark_detail_parser_matches_requested_command_shape():
args = build_parser().parse_args(
[
"benchmark",
"--name",
"SWE-Bench Pro",
"--limit",
"5",
"--require-metrics",
"Resolved,Latency",
"--min",
"Resolved=50",
"--max",
"Latency=2.5",
"--sort",
"Resolved:desc",
"--pareto",
"Resolved:higher,Latency:lower",
"--max-parameters",
"3B",
]
)
assert args.name == "SWE-Bench Pro"
assert args.limit == 5
assert args.require_metrics == ("Resolved", "Latency")
assert args.minimum_metrics == [("Resolved", 50)]
assert args.maximum_metrics == [("Latency", 2.5)]
assert args.sort_metric == ("Resolved", "desc")
assert args.pareto == (("Resolved", "higher"), ("Latency", "lower"))
assert args.max_parameters == 3_000_000_000
def test_benchmark_detail_parser_accepts_human_parameter_sizes():
parser = build_parser()
assert (
parser.parse_args(
["benchmark", "--name", "LM", "--max-parameters", "500m"]
).max_parameters
== 500_000_000
)
assert (
parser.parse_args(
["benchmark", "--name", "LM", "--max-parameters", "1.5B"]
).max_parameters
== 1_500_000_000
)
assert (
parser.parse_args(
["benchmark", "--name", "LM", "--max-parameters", "3_000_000_000"]
).max_parameters
== 3_000_000_000
)
for invalid in (
"0",
"-1",
"1.5",
"3K",
"3B parameters",
"_3B",
"3_B",
"3B_",
"1__000",
"1_.5B",
):
errors = io.StringIO()
with redirect_stderr(errors):
try:
parser.parse_args(
["benchmark", "--name", "LM", "--max-parameters", invalid]
)
except SystemExit as error:
assert error.code == 2
else:
raise AssertionError(f"accepted invalid parameter size: {invalid}")
assert "--max-parameters" in errors.getvalue()
def test_benchmark_list_rejects_parent_parameter_filter(monkeypatch, capsys):
class Client:
def __init__(self):
pass
def get(self, _path, _params=None):
return Response(b'{"count":0,"results":[]}', {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["benchmark", "--max-parameters", "3B", "list"]) == 2
assert capsys.readouterr().err == (
"pwc: --max-parameters is only valid with benchmark --name\n"
)
def test_task_benchmark_list_uses_frontend_ranking_and_aligned_output(monkeypatch):
calls = []
payload = {
"task_id": "18",
"task_slug": "ocr",
"recent_days": 90,
"results": [
{
"id": "2",
"slug": "recent-ocr",
"name": "Recent OCR",
"description": "Recent OCR benchmark.",
"hf_url": "https://huggingface.co/datasets/example/recent-ocr",
"recent_paper_count": 4,
"all_time_paper_count": 5,
"trend_score": 1.3333,
"support_weighted_score": 3.8627,
"best_model_name": "Reader 2",
"best_paper_title": "Recent OCR Paper",
"best_code_url": "https://github.com/example/reader-2",
},
{
"id": "1",
"slug": "classic-ocr",
"name": "Classic OCR",
"description": None,
"hf_url": None,
"recent_paper_count": 1,
"all_time_paper_count": 20,
"trend_score": 0.3333,
"support_weighted_score": 0.5941,
"best_model_name": "Reader 1",
"best_paper_title": "Classic OCR Paper",
"best_code_url": None,
},
],
}
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["benchmark", "list", "--task", "OCR"]) == 0
assert calls[0][0] == "tasks/OCR/trending-benchmarks"
assert calls[0][1]["min_recent_papers"] == 0
rows = output.getvalue().splitlines()
assert rows[0].split() == [
"id",
"slug",
"name",
"description",
"hf_url",
"recent_papers",
"ranking_score",
"best_model",
"paper_title",
"code",
]
assert "Recent OCR benchmark." in rows[1]
assert "https://huggingface.co/datasets/example/recent-ocr" in rows[1]
assert "Classic OCR" in rows[2]
def test_task_filter_takes_precedence_over_grouped_benchmark_display(
monkeypatch, capsys
):
payload = {
"count": 1,
"results": [
{
"id": "2",
"slug": "recent-ocr",
"name": "Recent OCR",
"recent_paper_count": 4,
"trend_score": 1.3333,
"best_model_name": "Reader 2",
"best_paper_title": "Recent OCR Paper",
}
],
}
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["benchmark", "list", "--task", "OCR", "--group-by-area"]) == 0
captured = capsys.readouterr()
assert calls[0][0] == "tasks/OCR/trending-benchmarks"
assert captured.out.startswith("id slug name")
assert captured.err == (
"# filters select task-scoped flat output; --group-by-area ignored\n"
)
def test_flat_benchmark_list_aligns_captured_column_widths(monkeypatch, capsys):
payload = {
"count": 2,
"results": [
{
"id": "2",
"name": "Short",
"description": "Short description",
"hf_url": None,
"full_name": "S",
"paper_count": 3,
},
{
"id": "10",
"name": "Longer Name",
"description": "Longer benchmark description",
"hf_url": "https://huggingface.co/datasets/example/longer",
"full_name": "Full benchmark name",
"paper_count": 12,
},
],
}
class Client:
def __init__(self):
pass
def get(self, _path, _params):
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["benchmark", "list", "--flat"]) == 0
rows = capsys.readouterr().out.splitlines()
assert rows[0].split() == [
"id",
"name",
"description",
"hf_url",
"full_name",
"evals",
]
assert "Short description" in rows[1]
assert "Longer benchmark description" in rows[2]
assert "https://huggingface.co/datasets/example/longer" in rows[2]
def test_benchmark_list_groups_top_benchmarks_as_markdown(monkeypatch):
calls = []
payload = {
"results": [
{
"id": "1",
"name": "Vision",
"tasks": [
{
"id": "10",
"name": "Image Classification",
"slug": "image-classification",
"paper_count": 100,
"benchmarks": [
{
"id": "20",
"name": "ImageNet [1K]",
"slug": "imagenet",
"full_name": "ImageNet Large Scale Visual Recognition",
"description": "Image classification benchmark.",
"hf_url": "https://huggingface.co/datasets/imagenet-1k",
"evaluation_count": 1234,
"featured_rank": 1,
}
],
}
],
}
]
}
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert (
main(
[
"benchmark",
"list",
"--group-by-area",
"--area",
"Vision",
"--benchmarks-per-task",
"5",
]
)
== 0
)
assert calls == [
(
"areas/with-task-benchmarks/",
{"benchmarks_per_task": 5, "min_eval_count": 2},
)
]
assert output.getvalue() == (
"# Area: Vision\n"
"\n"
"Inspect a benchmark with `pwc benchmark --name SLUG`.\n"
"\n"
"## Image Classification (`image-classification`)\n"
"\n"
"- ImageNet [1K] — description: Image classification benchmark.; HF: "
"https://huggingface.co/datasets/imagenet-1k; full name: ImageNet Large Scale Visual Recognition; "
"slug: `imagenet`; ID: `20`; 1,234 evaluations\n"
)
def test_task_list_resolves_area_name_and_renders_compact_output(monkeypatch):
calls = []
areas = {
"count": 2,
"results": [{"id": "1", "name": "Vision"}, {"id": "6", "name": "Audio"}],
}
tasks = {
"count": 1,
"results": [
{
"id": "10",
"slug": "image-classification",
"name": "Image Classification",
"area_id": "1",
"level": 0,
"paper_count": 123,
"benchmark_count": 4,
"evaluation_count": 567,
}
],
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
payload = areas if path == "areas/" else tasks
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert (
main(
[
"task",
"list",
"--area",
"vision",
"--search",
"image",
"--page-size",
"1",
"--order-by",
"paper_count",
"--order-dir",
"desc",
]
)
== 0
)
assert calls == [
("areas/", {"page": 1, "page_size": 500, "ordering": "name"}),
(
"tasks/",
{
"page": 1,
"page_size": 1,
"q": "image",
"area_id": "1",
"level": None,
"visible_only": False,
"ordering": "-paper_count",
},
),
]
assert output.getvalue() == (
"id\tslug\tname\tarea\tlevel\tpapers\tbenchmarks\tevals\n"
"10\timage-classification\tImage Classification\tVision\t0\t123\t4\t567\n"
)
def test_task_list_groups_complete_taxonomy_in_interactive_terminal(monkeypatch):
payload = {
"results": [
{
"id": "6",
"name": "Audio",
"tasks": [
{
"id": "259",
"slug": "audio-classification",
"name": "Audio Classification",
"paper_count": 63,
"benchmark_count": 1,
"evaluation_count": 8,
}
],
},
{
"id": "1",
"name": "Vision",
"tasks": [
{
"id": "13",
"slug": "3d-generation",
"name": "3D generation",
"paper_count": 2389,
"benchmark_count": 12,
"evaluation_count": 345,
},
{
"id": "1",
"slug": "image-classification",
"name": "Image Classification",
"paper_count": 1,
"benchmark_count": 1,
"evaluation_count": 1,
},
],
},
]
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert (path, params) == ("areas/with-tasks/", None)
return Response(json.dumps(payload).encode(), {})
class TerminalOutput(io.StringIO):
def isatty(self):
return True
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = TerminalOutput()
with redirect_stdout(output):
assert main(["task", "list"]) == 0
assert output.getvalue() == (
"# Area: Audio\n"
"- **Audio Classification** (`audio-classification`) — "
"63 papers · 1 benchmark · 8 evals\n"
"\n"
"# Area: Vision\n"
"- **3D generation** (`3d-generation`) — "
"2,389 papers · 12 benchmarks · 345 evals\n"
"- **Image Classification** (`image-classification`) — "
"1 paper · 1 benchmark · 1 eval\n"
)
assert "(`13`)" not in output.getvalue()
def test_task_list_can_force_grouped_area_output_when_captured(monkeypatch):
payload = {
"results": [
{"id": "6", "name": "Audio", "tasks": []},
{
"id": "1",
"name": "Vision",
"tasks": [
{
"id": "13",
"slug": "3d-generation",
"name": "3D generation",
"paper_count": 2389,
"benchmark_count": 12,
"evaluation_count": 345,
}
],
},
]
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert (path, params) == ("areas/with-tasks/", None)
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["task", "list", "--group-by-area", "--area", "vision"]) == 0
assert output.getvalue() == (
"# Area: Vision\n- **3D generation** (`3d-generation`) — "
"2,389 papers · 12 benchmarks · 345 evals\n"
)
def test_task_detail_renders_task_page_sections(monkeypatch):
calls = []
search = {
"count": 1,
"results": [
{
"id": "17",
"slug": "scene-text-recognition",
"name": "Scene Text Recognition",
"area_id": "1",
"level": 1,
"paper_count": 204,
"benchmark_count": 21,
"evaluation_count": 401,
}
],
}
page = {
"task": {
"id": "17",
"slug": "scene-text-recognition",
"name": "Scene Text Recognition",
"description": "Recognizing text in natural images.",
"research_trends": ["Vision-language pretraining"],
"area_id": "1",
"level": 1,
"paper_count": 204,
},
"parents": [{"id": "2", "slug": "ocr", "name": "OCR"}],
"sisters": [{"id": "18", "slug": "text-detection", "name": "Text Detection"}],
"recommended_frameworks": [
{
"id": "44",
"url": "https://github.com/huggingface/transformers",
"owner": "huggingface",
"name": "transformers",
"num_stars": 150000,
}
],
"children": [{"id": "19", "slug": "handwritten-text", "name": "Handwriting"}],
"benchmarks": [
{
"id": "30",
"slug": "iiit5k",
"name": "IIIT5K",
"paper_count": 12,
"best_model_name": "Model A",
},
{
"id": "32",
"slug": "svt",
"name": "SVT",
"paper_count": 4,
"best_model_name": "Model B",
},
],
"subtask_benchmarks": [
{
"subtask_id": "19",
"benchmarks": [{"id": "31", "slug": "iam", "name": "IAM"}],
}
],
}
papers = {
"count": 204,
"results": [
{
"id": "100",
"arxiv_id": "2501.00001",
"title": "A Better Text Recognizer",
"published": "2025-01-01",
"citation_count": 10,
"methods": [{"id": "7", "slug": "transformer", "name": "Transformer"}],
}
],
}
benchmark_trends = {
"support_prior_papers": 2.0,
"results": [
{
"id": "32",
"slug": "svt",
"name": "SVT",
"support_weighted_score": 2.5,
},
{
"id": "30",
"slug": "iiit5k",
"name": "IIIT5K",
"support_weighted_score": 1.25,
},
],
}
areas = {"count": 1, "results": [{"id": "1", "name": "Vision"}]}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
payloads = {
"tasks/": search,
"tasks/17/page": page,
"tasks/17/trending-benchmarks": benchmark_trends,
"tasks/17/papers": papers,
"areas/": areas,
}
return Response(json.dumps(payloads[path]).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["task", "--name", "scene-text-recognition"]) == 0
rendered = output.getvalue()
assert "# Scene Text Recognition" in rendered
assert (
"Area: Vision · Level: 1 · 204 papers · 21 benchmarks · 401 evaluations"
in rendered
)
assert "## Recommended frameworks" in rendered
assert (
"[huggingface/transformers](https://github.com/huggingface/transformers)"
in rendered
)
assert "## Sister tasks" in rendered
assert "## Common methods (from 1 sampled papers)" in rendered
assert "## Benchmarks by subtask" in rendered
assert "## Trending papers (top 1 of 204)" in rendered
assert rendered.index("SVT") < rendered.index("IIIT5K")
assert "trend score: 2.50" in rendered
assert calls == [
("tasks/", {"q": "scene-text-recognition", "page": 1, "page_size": 100}),
("tasks/17/page", None),
(
"tasks/17/trending-benchmarks",
{"limit": 100, "min_recent_papers": 0},
),
(
"tasks/17/papers",
{
"page": 1,
"page_size": 100,
"order_by": "trending",
"order_dir": "desc",
"include_resources": True,
},
),
("areas/", {"page": 1, "page_size": 500, "ordering": "name"}),
]
def test_task_detail_json_has_stable_composed_payload(monkeypatch, capsys):
payloads = {
"tasks/": {
"count": 1,
"results": [
{
"id": "17",
"slug": "scene-text-recognition",
"name": "Scene Text Recognition",
"area_id": "1",
}
],
},
"tasks/17/page": {
"task": {"id": "17", "name": "Scene Text Recognition"},
"recommended_frameworks": [],
"benchmarks": [],
},
"tasks/17/trending-benchmarks": {"results": []},
"tasks/17/papers": {"count": 0, "results": []},
"tasks/17": {
"id": "17",
"slug": "scene-text-recognition",
"name": "Scene Text Recognition",
"area_id": "1",
},
"areas/": {"count": 1, "results": [{"id": "1", "name": "Vision"}]},
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
return Response(json.dumps(payloads[path]).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["task", "--name", "Scene Text Recognition", "--json"]) == 0
result = json.loads(capsys.readouterr().out)
assert result["schema_version"] == "v1"
assert result["data"]["task"]["id"] == "17"
assert result["data"]["area"]["name"] == "Vision"
assert result["data"]["paper_sample_size"] == 0
assert main(["task", "--name", "17", "--json"]) == 0
result = json.loads(capsys.readouterr().out)
assert result["data"]["task"]["id"] == "17"
def test_method_list_accepts_area_id_and_year_filter(monkeypatch):
calls = []
areas = {
"count": 1,
"results": [{"id": "6", "name": "Audio"}],
}
methods = {
"count": 1,
"results": [
{
"id": "20",
"slug": "wav2vec-2",
"name": "wav2vec 2.0",
"full_name": "wav2vec 2.0",
"area_id": "6",
"introduced_year": 2020,
"paper_count": 45,
}
],
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
payload = areas if path == "areas/" else methods
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert (
main(
[
"method",
"list",
"--area",
"6",
"--search",
"wav2vec",
"--introduced-year",
"2020",
]
)
== 0
)
assert calls[1] == (
"methods/",
{
"page": 1,
"page_size": 50,
"q": "wav2vec",
"area_id": "6",
"introduced_year": 2020,
"ordering": "name",
},
)
assert output.getvalue() == (
"id\tslug\tname\tfull_name\tarea\tintroduced\tpapers\n"
"20\twav2vec-2\twav2vec 2.0\twav2vec 2.0\tAudio\t2020\t45\n"
)
def test_method_detail_resolves_exact_name_and_renders_source(monkeypatch):
calls = []
payloads = {
"methods/": {
"count": 1,
"results": [
{
"id": "2",
"slug": "transformer",
"name": "Transformer",
"full_name": "Transformer",
}
],
},
"methods/2": {
"id": "2",
"slug": "transformer",
"name": "Transformer",
"description": "Attention-only architecture.",
"area_id": "3",
"introduced_year": 2017,
"paper_count": 13086,
"source_url": "/paper/1706.03762",
"source_title": "Attention Is All You Need",
},
"areas/": {"count": 1, "results": [{"id": "3", "name": "General"}]},
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
return Response(json.dumps(payloads[path]).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["method", "--name", "Transformer"]) == 0
rendered = output.getvalue()
assert "# Transformer" in rendered
assert "Area: General · Introduced: 2017 · 13,086 papers" in rendered
assert (
"[Attention Is All You Need](https://paperswithcode.co/paper/1706.03762)"
in rendered
)
assert calls[0] == (
"methods/",
{"q": "Transformer", "page": 1, "page_size": 100},
)
calls.clear()
with redirect_stdout(io.StringIO()):
assert main(["method", "--name", "2"]) == 0
assert calls == [
("methods/2", None),
("areas/", {"page": 1, "page_size": 500, "ordering": "name"}),
]
def _keyword_method_client(monkeypatch, catalog):
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
if path == "methods/" and params.get("q"):
payload = {"count": 0, "results": []}
elif path == "methods/":
payload = {"count": len(catalog), "results": catalog}
elif path == "areas/":
payload = {"count": 0, "results": []}
else:
method_id = path.split("/")[1]
payload = next(m for m in catalog if m["id"] == method_id)
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
return calls
def test_method_detail_resolves_keyword_alias(monkeypatch):
catalog = [
{
"id": "1205",
"slug": "gbdt",
"name": "Gradient Boosted Decision Trees",
"full_name": "Gradient Boosted Decision Trees (GBDT)",
"keywords": "GBDT, GBDTs, GDBT, gradient boosting machine",
},
{
"id": "9",
"slug": "pi0",
"name": "pi0",
"full_name": "pi0",
"keywords": "pi-0, pi zero",
},
]
calls = _keyword_method_client(monkeypatch, catalog)
for alias, slug in (("gdbt", "gbdt"), ("GDBT", "gbdt"), ("pi 0", "pi0")):
output = io.StringIO()
with redirect_stdout(output):
assert main(["method", "--name", alias, "--json"]) == 0
assert json.loads(output.getvalue())["data"]["method"]["slug"] == slug
assert ("methods/", {"page": 1, "page_size": 500}) in calls
def test_method_detail_keyword_alias_must_be_unique(monkeypatch, capsys):
catalog = [
{"id": "1", "slug": "a", "name": "Alpha", "keywords": "shared, one"},
{"id": "2", "slug": "b", "name": "Beta", "keywords": "Shared"},
]
_keyword_method_client(monkeypatch, catalog)
assert main(["method", "--name", "shared"]) != 0
assert "Method not found: shared" in capsys.readouterr().err
def test_conference_list_filters_complete_response_by_year(monkeypatch):
payload = {
"count": 2,
"results": [
{
"slug": "cvpr-2025",
"name": "CVPR 2025",
"year": 2025,
"paper_count": 1834,
"location": "Nashville, USA",
},
{
"slug": "cvpr-2026",
"name": "CVPR 2026",
"year": 2026,
"paper_count": 2200,
"location": "Denver, USA",
},
],
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert (path, params) == ("conferences/", None)
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["conference", "list", "--year", "2025"]) == 0
assert output.getvalue() == (
"slug\tname\tyear\tpapers\tlocation\n"
"cvpr-2025\tCVPR 2025\t2025\t1834\tNashville, USA\n"
)
def test_conference_detail_accepts_name_slug_or_id(monkeypatch, capsys):
listing = {
"count": 1,
"results": [
{"id": "7", "slug": "cvpr-2026", "name": "CVPR 2026", "year": 2026}
],
}
detail = {
"id": "7",
"slug": "cvpr-2026",
"name": "CVPR 2026",
"year": 2026,
"paper_count": 2677,
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
payload = listing if path == "conferences/" else detail
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
for reference in ("CVPR 2026", "cvpr-2026", "7"):
assert main(["conference", "--name", reference, "--json"]) == 0
result = json.loads(capsys.readouterr().out)
assert result["data"]["slug"] == "cvpr-2026"
def test_conference_list_aligns_columns_in_interactive_terminal(monkeypatch):
payload = {
"count": 2,
"results": [
{
"slug": "cvpr-2025",
"name": "CVPR 2025",
"year": 2025,
"paper_count": 1834,
"location": "Nashville, USA",
},
{
"slug": "neurips-2025",
"name": "NeurIPS 2025",
"year": 2025,
"paper_count": 2763,
"location": "San Diego, USA",
},
],
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert (path, params) == ("conferences/", None)
return Response(json.dumps(payload).encode(), {})
class TerminalOutput(io.StringIO):
def isatty(self):
return True
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = TerminalOutput()
with redirect_stdout(output):
assert main(["conference", "list"]) == 0
assert output.getvalue() == (
"slug name year papers location\n"
"cvpr-2025 CVPR 2025 2025 1834 Nashville, USA\n"
"neurips-2025 NeurIPS 2025 2025 2763 San Diego, USA\n"
)
def test_organization_detail_and_featured_list(monkeypatch, capsys):
listing = {
"count": 1,
"results": [
{
"id": "1",
"slug": "nvidia",
"name": "NVIDIA",
"paper_count": 158,
"is_featured": True,
"trending_rank": 1,
}
],
}
detail = {
**listing["results"][0],
"hf_url": "https://huggingface.co/nvidia",
"website_url": "https://www.nvidia.com/",
}
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
payload = detail if path == "organizations/nvidia" else listing
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["organization", "--name", "NVIDIA", "--json"]) == 0
result = json.loads(capsys.readouterr().out)
assert result["data"]["slug"] == "nvidia"
assert main(["organization", "list", "--featured-only"]) == 0
assert "1\tnvidia\tNVIDIA\t158\tyes\t1" in capsys.readouterr().out
assert calls == [
("organizations/", {"featured_only": False}),
("organizations/nvidia", None),
("organizations/", {"featured_only": True}),
]
def test_framework_detail_and_filtered_list_flatten_catalog(monkeypatch, capsys):
catalog = {
"domains": [
{
"slug": "llm",
"name": "Large Language Models",
"categories": [
{
"slug": "inference",
"name": "Inference & Serving",
"frameworks": [
{
"id": "11",
"slug": "vllm",
"name": "vLLM",
"description": "High-throughput serving.",
"platforms": ["gpu"],
"paper_count": 73,
},
{
"id": "9",
"slug": "llama-cpp",
"name": "llama.cpp",
"description": "Portable inference.",
"platforms": ["cpu", "apple"],
"paper_count": 2,
},
],
}
],
}
]
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert (path, params) == ("frameworks/", None)
return Response(json.dumps(catalog).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["framework", "--name", "11", "--json"]) == 0
result = json.loads(capsys.readouterr().out)
assert result["data"]["slug"] == "vllm"
assert result["data"]["domain_slug"] == "llm"
assert main(["framework", "list", "--platform", "apple"]) == 0
output = capsys.readouterr().out
assert "llama-cpp" in output
assert "vllm" not in output
def test_unknown_area_lists_valid_choices(monkeypatch):
areas = {
"count": 2,
"results": [{"id": "1", "name": "Vision"}, {"id": "6", "name": "Audio"}],
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert path == "areas/"
return Response(json.dumps(areas).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
errors = io.StringIO()
with redirect_stderr(errors):
assert main(["task", "list", "--area", "Robotics"]) == 4
assert errors.getvalue() == (
"pwc: Area not found: Robotics; available areas: Audio, Vision\n"
)
def test_benchmark_detail_renders_merged_markdown_leaderboard(monkeypatch):
calls = []
benchmark_payload = {
"count": 1,
"results": [
{
"id": "42",
"name": "SWE-Bench Pro",
"slug": "swe-bench-pro",
"description": "A coding-agent benchmark.",
}
],
}
evaluations_payload = {
"count": 3,
"results": [
{
"id": "1",
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Agent | One",
"metrics": {"Resolved": 55.5},
"best_metric": "Resolved",
"best_rank": 1,
"num_parameters": 10_000_000_000,
"paper_title": "An Agent Paper",
"paper_arxiv_id": "2601.12345",
"paper_published_date": "2026-01-20",
"task_name": "Coding Agents",
"is_open": False,
},
{
"id": "2",
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Agent | One",
"metrics": {"Pass@1": 50},
"best_metric": "Pass@1",
"best_rank": 2,
"num_parameters": 10_000_000_000,
"paper_title": "An Agent Paper",
"paper_arxiv_id": "2601.12345",
"paper_published_date": "2026-01-20",
"task_name": "Coding Agents",
"is_open": False,
},
{
"id": "3",
"paper_id": "9",
"task_id": "8",
"dataset_id": "42",
"model_name": "Agent | One",
"metrics": {"Resolved": 55.5},
"best_metric": "Resolved",
"best_rank": 4,
"num_parameters": 10_000_000_000,
"paper_title": "An Agent Paper",
"paper_arxiv_id": "2601.12345",
"paper_published_date": "2026-01-20",
"task_name": "Software Engineering",
"task_slug": "software-engineering",
"is_open": False,
},
],
}
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
payload = benchmark_payload if path == "datasets/" else evaluations_payload
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["benchmark", "--name", "SWE-Bench Pro"]) == 0
rendered = output.getvalue()
assert "# SWE-Bench Pro" in rendered
assert "Agent \\| One" in rendered
assert "| 1 | Agent \\| One | 10B |" in rendered
assert "Resolved: 55.5, Pass@1: 50" in rendered
assert "[An Agent Paper](https://paperswithcode.co/paper/2601.12345)" in rendered
assert "2026-01-20" in rendered
assert calls[1] == (
"datasets/42/evaluations/",
{
"page": 1,
"page_size": 20,
"ordering": "best_rank",
"is_open": None,
},
)
json_output = io.StringIO()
with redirect_stdout(json_output):
assert main(["benchmark", "--name", "SWE-Bench Pro", "--json"]) == 0
rows = json.loads(json_output.getvalue())["data"]["results"]
assert len(rows) == 1
assert rows[0]["rank_scopes"] == [
{
"task_id": "3",
"task_name": "Coding Agents",
"task_slug": None,
"rank": 1,
},
{
"task_id": "8",
"task_name": "Software Engineering",
"task_slug": "software-engineering",
"rank": 4,
},
]
def test_benchmark_detail_renders_aligned_table_in_terminal(monkeypatch):
benchmark_payload = {
"count": 1,
"results": [
{
"id": "42",
"name": "SWE-Bench Pro",
"slug": "swe-bench-pro",
"description": "A coding-agent benchmark.",
}
],
}
evaluations_payload = {
"count": 1,
"results": [
{
"id": "1",
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Agent One",
"metrics": {"Resolved": 55.5},
"best_metric": "Resolved",
"best_rank": 1,
"paper_title": "An Agent Paper",
"paper_arxiv_id": "2601.12345",
"paper_published_date": "2026-01-20",
"task_name": "Coding Agents",
"is_open": False,
}
],
}
class Client:
def __init__(self):
pass
def get(self, path, params):
payload = benchmark_payload if path == "datasets/" else evaluations_payload
return Response(json.dumps(payload).encode(), {})
class TerminalBuffer(io.StringIO):
def isatty(self):
return True
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = TerminalBuffer()
with redirect_stdout(output):
assert main(["benchmark", "--name", "SWE-Bench Pro"]) == 0
rendered = output.getvalue()
assert rendered.startswith("SWE-Bench Pro\n\nA coding-agent benchmark.\n")
assert (
"View benchmark: https://paperswithcode.co/benchmark/swe-bench-pro" in rendered
)
assert "Rank Model Parameters Scores Task" in rendered
assert " 1 Agent One — Resolved: 55.5 Coding Agents" in rendered
assert "An Agent Paper [2601.12345]" in rendered
assert "| ---:" not in rendered
def _parameter_filter_client(monkeypatch, evaluations_payload):
benchmark_payload = {
"count": 1,
"results": [{"id": "42", "name": "Language Model Benchmark", "slug": "lmb"}],
}
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
payload = benchmark_payload if path == "datasets/" else evaluations_payload
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
return calls
def test_benchmark_detail_filters_by_inclusive_max_parameters(monkeypatch, capsys):
evaluations_payload = {
"count": 1,
"next_page": None,
"previous_page": None,
"parameter_coverage_known": 1,
"parameter_coverage_total": 2,
"results": [
{
"id": "1",
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Edge 3B",
"metrics": {"Perplexity": 12.5},
"best_metric": "Perplexity",
"best_rank": 4,
"num_parameters": 3_000_000_000,
"is_open": True,
}
],
}
calls = _parameter_filter_client(monkeypatch, evaluations_payload)
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--is-open",
"true",
"--max-parameters",
"3B",
"--json",
]
)
== 0
)
assert calls[1] == (
"evaluations/",
{
"page": 1,
"page_size": 20,
"dataset_id": "42",
"ordering": "best_rank",
"is_open": "true",
"max_parameters_exclusive": 3_000_000_001,
},
)
result = json.loads(capsys.readouterr().out)
assert result["data"]["matched_count"] == 1
assert result["data"]["results"][0]["model_name"] == "Edge 3B"
assert result["data"]["results"][0]["num_parameters"] == 3_000_000_000
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--max-parameters",
"3B",
]
)
== 0
)
rendered = capsys.readouterr().out
assert "| Rank | Model | Parameters | Scores |" in rendered
assert "| 4 | Edge 3B | 3B | Perplexity: 12.5 |" in rendered
def test_benchmark_detail_parameter_filter_fails_closed_without_api_support(
monkeypatch, capsys
):
unsupported_payload = {"count": 0, "results": []}
_parameter_filter_client(monkeypatch, unsupported_payload)
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--max-parameters",
"3B",
]
)
== 4
)
assert capsys.readouterr().err == (
"pwc: parameter filtering requires a compatible API; server must be upgraded\n"
)
def test_benchmark_detail_parameter_filter_rejects_out_of_bound_response(
monkeypatch, capsys
):
invalid_payload = {
"count": 1,
"parameter_coverage_known": 1,
"parameter_coverage_total": 1,
"results": [
{
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Too Large",
"metrics": {"Perplexity": 10},
"num_parameters": 3_000_000_001,
}
],
}
_parameter_filter_client(monkeypatch, invalid_payload)
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--max-parameters",
"3B",
]
)
== 4
)
assert capsys.readouterr().err == (
"pwc: API returned a model outside the requested parameter limit\n"
)
def test_benchmark_detail_parameter_filter_rejects_conflicting_sibling_counts(
monkeypatch, capsys
):
invalid_payload = {
"count": 1,
"parameter_coverage_known": 1,
"parameter_coverage_total": 1,
"results": [
{
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Conflicting",
"metrics": {"Perplexity": 10},
"num_parameters": 2_000_000_000,
},
{
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Conflicting",
"metrics": {"Accuracy": 90},
"num_parameters": 2_500_000_000,
},
],
}
_parameter_filter_client(monkeypatch, invalid_payload)
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--max-parameters",
"3B",
]
)
== 4
)
assert capsys.readouterr().err == (
"pwc: API returned a model outside the requested parameter limit\n"
)
def test_benchmark_detail_parameter_filter_rejects_incomplete_sibling_counts(
monkeypatch, capsys
):
invalid_payload = {
"count": 1,
"parameter_coverage_known": 1,
"parameter_coverage_total": 1,
"results": [
{
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Incomplete",
"metrics": {"Perplexity": 10},
"num_parameters": 2_000_000_000,
},
{
"paper_id": "9",
"task_id": "3",
"dataset_id": "42",
"model_name": "Incomplete",
"metrics": {"Accuracy": 90},
"num_parameters": None,
},
],
}
_parameter_filter_client(monkeypatch, invalid_payload)
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--max-parameters",
"3B",
]
)
== 4
)
assert capsys.readouterr().err == (
"pwc: API returned a model outside the requested parameter limit\n"
)
def test_benchmark_detail_parameter_filter_composes_with_metric_pagination(
monkeypatch, capsys
):
benchmark_payload = {
"count": 1,
"results": [{"id": "42", "name": "Language Model Benchmark"}],
}
pages = {
1: {
"count": 2,
"next_page": 2,
"parameter_coverage_known": 2,
"parameter_coverage_total": 3,
"results": [
{
"paper_id": "1",
"task_id": "3",
"dataset_id": "42",
"model_name": "Model A",
"metrics": {"Accuracy": 80},
"num_parameters": 1_000_000_000,
},
{
"paper_id": "1",
"task_id": "3",
"dataset_id": "42",
"model_name": "Model A",
"metrics": {"Latency": 5},
"num_parameters": 1_000_000_000,
},
],
},
2: {
"count": 2,
"next_page": None,
"parameter_coverage_known": 2,
"parameter_coverage_total": 3,
"results": [
{
"paper_id": "2",
"task_id": "3",
"dataset_id": "42",
"model_name": "Model B",
"metrics": {"Accuracy": 90, "Latency": 7},
"num_parameters": 2_000_000_000,
}
],
},
}
evaluation_calls = []
class Client:
def __init__(self):
pass
def get(self, path, params):
if path == "datasets/":
return Response(json.dumps(benchmark_payload).encode(), {})
evaluation_calls.append(params)
return Response(json.dumps(pages[params["page"]]).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert (
main(
[
"benchmark",
"--name",
"Language Model Benchmark",
"--max-parameters",
"2B",
"--require-metrics",
"Accuracy,Latency",
"--sort",
"Accuracy:desc",
"--json",
]
)
== 0
)
result = json.loads(capsys.readouterr().out)
assert [row["model_name"] for row in result["data"]["results"]] == [
"Model B",
"Model A",
]
assert [call["page"] for call in evaluation_calls] == [1, 2]
def test_benchmark_detail_filters_sorts_and_computes_pareto_frontier(
monkeypatch, capsys
):
benchmark_payload = {
"count": 1,
"results": [{"id": "6", "name": "COCO", "slug": "coco"}],
}
evaluation_pages = {
1: {
"count": 3,
"results": [
{
"id": "1",
"paper_id": "1",
"task_id": "3",
"dataset_id": "6",
"model_name": "Accurate",
"metrics": {"mAP": 60, "FPS": 30},
"best_rank": 1,
},
{
"id": "2",
"paper_id": "2",
"task_id": "3",
"dataset_id": "6",
"model_name": "Fast",
"metrics": {"mAP": "58", "FPS": "100"},
"best_rank": 2,
},
],
},
2: {
"count": 3,
"results": [
{
"id": "3",
"paper_id": "3",
"task_id": "3",
"dataset_id": "6",
"model_name": "Dominated",
"metrics": {"mAP": 57, "FPS": 50},
"best_rank": 3,
}
],
},
}
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
if path == "datasets/":
return Response(json.dumps(benchmark_payload).encode(), {})
return Response(json.dumps(evaluation_pages[params["page"]]).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert (
main(
[
"benchmark",
"--name",
"COCO",
"--pareto",
"mAP:higher,FPS:higher",
"--sort",
"FPS:desc",
]
)
== 0
)
captured = capsys.readouterr()
assert captured.err == ""
assert (
"Showing 2 of 2 matching merged rows from 3 stored evaluation rows."
in captured.out
)
assert captured.out.index("Fast") < captured.out.index("Accurate")
assert "Dominated" not in captured.out
assert [params["page"] for path, params in calls if path != "datasets/"] == [1, 2]
def test_benchmark_detail_metric_thresholds_and_unknown_metric(monkeypatch, capsys):
benchmark_payload = {
"count": 1,
"results": [{"id": "6", "name": "COCO", "slug": "coco"}],
}
evaluations_payload = {
"count": 2,
"results": [
{
"paper_id": "1",
"task_id": "3",
"dataset_id": "6",
"model_name": "Fast",
"metrics": {"mAP": 58, "FPS": 100},
},
{
"paper_id": "2",
"task_id": "3",
"dataset_id": "6",
"model_name": "Slow",
"metrics": {"mAP": 62, "FPS": 20},
},
],
}
class Client:
def __init__(self):
pass
def get(self, path, _params):
payload = benchmark_payload if path == "datasets/" else evaluations_payload
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert (
main(
[
"benchmark",
"--name",
"COCO",
"--require-metrics",
"mAP,FPS",
"--min",
"FPS=60",
"--max",
"mAP=60",
]
)
== 0
)
captured = capsys.readouterr()
assert "Fast" in captured.out
assert "Slow" not in captured.out
assert main(["benchmark", "--name", "COCO", "--sort", "Latency"]) == 2
captured = capsys.readouterr()
assert captured.out == ""
assert "unknown metric(s): latency; available metrics: FPS, mAP" in captured.err
def test_top_level_version_is_offline_and_stable():
output = io.StringIO()
with redirect_stdout(output):
try:
build_parser().parse_args(["--version"])
except SystemExit as error:
assert error.code == 0
assert output.getvalue() == "pwc 0.4.6\tapi v1\n"
def test_search_default_output_is_compact_deterministic_tsv(monkeypatch):
payload = {
"count": 1,
"results": [
{
"arxiv_id": "2501.01234",
"title": "Line one\nline two",
"published": "2025-01-03",
"citation_count": 42,
"has_official_implementation": True,
"code_repository_count": 3,
}
],
}
class Client:
def __init__(self):
pass
def get(self, _path, _params):
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["search", "vision"]) == 0
assert output.getvalue() == (
"id\ttitle\tyear\tcitations\n2501.01234\tLine one\\nline two\t2025\t42\n"
)
output = io.StringIO()
with redirect_stdout(output):
assert main(["search", "vision", "--implementation-coverage"]) == 0
assert output.getvalue() == (
"id\ttitle\tyear\tcitations\tofficial_implementation\tcode_repositories\n"
"2501.01234\tLine one\\nline two\t2025\t42\ttrue\t3\n"
)
def test_search_official_implementation_filter_is_forwarded_and_fail_closed(
monkeypatch, capsys
):
calls = []
payloads = [
{"results": [], "applied_filters": {}},
{
"results": [],
"applied_filters": {"has_official_implementation": "true"},
},
]
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
return Response(json.dumps(payloads.pop(0)).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
command = ["search", "vision", "--has-official-implementation"]
assert main(command) == 4
assert "did not confirm has_official_implementation" in capsys.readouterr().err
assert main(command) == 0
capsys.readouterr()
assert calls[0][0] == "papers/search"
assert calls[0][1]["has_official_implementation"] is True
def test_paper_list_forwards_confirmed_official_implementation_filter(monkeypatch):
calls = []
payload = {
"results": [],
"applied_filters": {"has_official_implementation": "true"},
}
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
assert main(["paper", "list", "--has-official-implementation"]) == 0
assert calls[0][0] == "papers/"
assert calls[0][1]["has_official_implementation"] is True
def test_all_paper_commands_resolve_exact_titles(monkeypatch):
title = "Transformers Are RNNs"
resolved = "2006.16236"
search_payload = {
"count": 1,
"results": [{"arxiv_id": resolved, "title": title}],
}
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
calls.append((path, params))
if path == "papers/search":
return Response(json.dumps(search_payload).encode(), {})
if path == f"papers/{resolved}":
return Response(
json.dumps({"arxiv_id": resolved, "title": title}).encode(), {}
)
if path == f"research/papers/{resolved}/read":
return Response(b"# Paper", {})
if path == f"papers/{resolved}/related":
return Response(b'{"count":0,"results":[]}', {})
if path == f"research/papers/{resolved}/lineage":
return Response(
json.dumps(
{
"paper": {"reference": resolved, "title": title},
"predecessors": [],
"successors": [],
}
).encode(),
{},
)
raise AssertionError(path)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
commands = (
(
["paper", "info", title.lower()],
(f"papers/{resolved}", {"include_resources": False}),
),
(
["paper", "read", title.lower()],
(f"research/papers/{resolved}/read", None),
),
(
["paper", "related", title.lower()],
(f"papers/{resolved}/related", {"limit": 4}),
),
(
["paper", "lineage", "list", title.lower()],
(f"research/papers/{resolved}/lineage", None),
),
)
for argv, expected_request in commands:
calls.clear()
with redirect_stdout(io.StringIO()):
assert main(argv) == 0
assert calls == [
(
"papers/search",
{
"q": title.lower(),
"page": 1,
"page_size": 20,
"mode": "keyword",
},
),
expected_request,
]
def test_ambiguous_paper_title_is_rejected_with_matches(monkeypatch):
title = "A Shared Title"
payload = {
"count": 2,
"results": [
{"arxiv_id": "2501.00001", "title": title},
{"arxiv_id": "2501.00002", "title": title},
],
}
class Client:
def __init__(self):
pass
def get(self, path, params=None):
assert path == "papers/search"
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
errors = io.StringIO()
with redirect_stderr(errors):
assert main(["paper", "info", title]) == 4
assert errors.getvalue() == (
"pwc: Paper title is ambiguous: A Shared Title; "
"A Shared Title (2501.00001); A Shared Title (2501.00002)\n"
)
def test_paper_info_renders_metadata_and_abstract_sections(monkeypatch):
payload = {
"arxiv_id": "2501.01234",
"title": "A Paper",
"abstract": "First line.\nSecond line.",
"published": "2025-01-03",
"authors": ["Ada Researcher", "Grace Scientist"],
"organizations": [{"id": "1", "name": "Example AI", "slug": "example"}],
"citation_count": 42,
"has_official_implementation": False,
"code_repository_count": 2,
"url_abs": "https://paperswithcode.co/paper/2501.01234",
}
class Client:
def __init__(self):
pass
def get(self, _path, _params):
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "info", "2501.01234"]) == 0
assert output.getvalue() == (
"id: 2501.01234\n"
"title: A Paper\n"
"published: 2025-01-03\n"
"authors: Ada Researcher, Grace Scientist\n"
"organizations: Example AI\n"
"citations: 42\n"
"official_implementation: false\n"
"code_repositories: 2\n"
"url: https://paperswithcode.co/paper/2501.01234\n"
"\n"
"## Abstract\n"
"\n"
"First line.\n"
"Second line.\n"
"\n"
"## Paper lineage\n"
"\n"
"### Predecessors\n"
"\n"
"- None\n"
"\n"
"### Successors\n"
"\n"
"- None\n"
)
def test_paper_info_renders_linked_lineage_sections(monkeypatch):
payload = {
"arxiv_id": "2501.01234",
"title": "A Paper",
"predecessor_papers": [
{
"title": "Earlier [Paper]",
"route_identifier": "2401.00001",
"published": "2024-01-02",
}
],
"successor_papers": [
{
"title": "Later Paper",
"route_identifier": "2601.00001",
"published": None,
}
],
}
class Client:
def __init__(self):
pass
def get(self, _path, _params):
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "info", "2501.01234"]) == 0
assert output.getvalue().endswith(
"\n## Paper lineage\n\n"
"### Predecessors\n\n"
"- [Earlier \\[Paper\\]](https://paperswithcode.co/paper/2401.00001)"
" (2024-01-02)\n\n"
"### Successors\n\n"
"- [Later Paper](https://paperswithcode.co/paper/2601.00001)\n"
)
def test_paper_info_renders_linked_resources_when_requested(monkeypatch):
calls = []
payload = {
"arxiv_id": "2501.01234",
"title": "A Paper",
"has_official_implementation": True,
"code_repository_count": 2,
"repositories": [
{"url": "https://github.com/example/implementation", "is_official": False},
{"url": "https://github.com/example/official", "is_official": True},
],
"project_pages": [
{"url": "https://example.org/community", "is_official": False},
{"url": "https://example.org/project", "is_official": True},
],
"hf_models": ["https://huggingface.co/example/model"],
"hf_datasets": ["https://huggingface.co/datasets/example/data"],
"hf_spaces": ["https://huggingface.co/spaces/example/demo"],
"x_threads": [
{
"url": "https://x.com/author/status/123",
"author_name": "Ada Author",
"author_handle": "author",
}
],
}
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "info", "2501.01234", "--include-resources"]) == 0
assert calls == [
("papers/2501.01234", {"include_resources": True}),
]
assert output.getvalue() == (
"id: 2501.01234\n"
"title: A Paper\n"
"official_implementation: true\n"
"code_repositories: 2\n"
"\n"
"## Paper lineage\n"
"\n"
"### Predecessors\n"
"\n"
"- None\n"
"\n"
"### Successors\n"
"\n"
"- None\n"
"\n"
"## Repositories\n"
"\n"
"- **Official:** https://github.com/example/official\n"
"- https://github.com/example/implementation\n"
"\n"
"## Project pages\n"
"\n"
"- **Official:** https://example.org/project\n"
"- https://example.org/community\n"
"\n"
"## Hugging Face artifacts\n"
"\n"
"- **Model:** https://huggingface.co/example/model\n"
"- **Dataset:** https://huggingface.co/datasets/example/data\n"
"- **Space:** https://huggingface.co/spaces/example/demo\n"
"\n"
"## X threads\n"
"\n"
"- **Ada Author (@author):** https://x.com/author/status/123\n"
)
def test_paper_info_include_evals_renders_paginated_markdown_table(monkeypatch):
calls = []
paper_payload = {
"id": "9",
"arxiv_id": "2607.24653",
"title": "An Evaluated Paper",
}
evaluation_rows = [
{
"id": "1",
"paper_id": "9",
"model_name": "Agent | One",
"metrics": {"Resolved": 55.5},
"best_metric": "Resolved",
"best_rank": 1,
"task_name": "Coding Agents",
"dataset_name": "SWE-Bench Pro",
"is_open": True,
"result_url": "https://example.test/result-one",
},
{
"id": "2",
"paper_id": "9",
"model_name": "Agent Two",
"harness": "verified",
"metrics": {"Accuracy": 50},
"best_metric": "Accuracy",
"best_rank": None,
"task_name": "Agents",
"dataset_name": "Toolathlon",
"is_open": False,
"source_url": "https://example.test/result-two",
},
]
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
if path.startswith("papers/"):
payload = paper_payload
else:
payload = {
"count": 2,
"results": [evaluation_rows[params["page"] - 1]],
}
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "info", "2607.24653", "--include-evals"]) == 0
assert calls == [
("papers/2607.24653", {"include_resources": False}),
(
"evaluations/",
{
"page": 1,
"page_size": 100,
"paper_id": "9",
"ordering": "-benchmark_popularity",
},
),
(
"evaluations/",
{
"page": 2,
"page_size": 100,
"paper_id": "9",
"ordering": "-benchmark_popularity",
},
),
]
rendered = output.getvalue()
assert "\n## Evaluations\n" in rendered
assert "| Benchmark | Model | Scores | Rank | Task | Open | Source |" in rendered
assert "Agent \\| One" in rendered
assert "Resolved: 55.5" in rendered
assert "Agent Two (verified)" in rendered
assert (
"| — | Agents | no | [source](<https://example.test/result-two>) |" in rendered
)
def test_paper_info_include_evals_adds_structured_json(monkeypatch):
calls = []
class Client:
def __init__(self):
pass
def get(self, path, params):
calls.append((path, params))
payload = (
{"id": "9", "arxiv_id": "2607.24653", "title": "Paper"}
if path.startswith("papers/")
else {"count": 0, "results": []}
)
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "info", "2607.24653", "--include-evals", "--json"]) == 0
data = json.loads(output.getvalue())["data"]
assert data["evaluations"] == {"count": 0, "results": []}
assert calls[-1][0] == "evaluations/"
def test_paper_lineage_renders_linked_markdown_sections(monkeypatch):
payload = {
"paper": {
"reference": "2307.09288",
"title": "Llama 2: Open Foundation and Fine-Tuned Chat Models",
},
"predecessors": [
{
"reference": "2302.13971",
"title": "LLaMA: Open and Efficient Foundation Language Models",
}
],
"successors": [],
}
class Client:
def __init__(self):
pass
def get(self, path):
assert path == "research/papers/2307.09288/lineage"
return Response(json.dumps(payload).encode(), {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "lineage", "list", "2307.09288"]) == 0
assert output.getvalue() == (
"# Paper lineage\n"
"\n"
"## Paper\n"
"\n"
"- [Llama 2: Open Foundation and Fine-Tuned Chat Models]"
"(https://paperswithcode.co/paper/2307.09288) (`2307.09288`)\n"
"\n"
"## Predecessors\n"
"\n"
"- [LLaMA: Open and Efficient Foundation Language Models]"
"(https://paperswithcode.co/paper/2302.13971) (`2302.13971`)\n"
"\n"
"## Successors\n"
"\n"
"- None\n"
)
def test_json_schema_and_errors_use_stable_streams(monkeypatch):
class Client:
def __init__(self):
pass
def get(self, _path, _params):
return Response(b'{"count":0,"results":[]}', {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
errors = io.StringIO()
with redirect_stdout(output), redirect_stderr(errors):
assert main(["search", "none", "--json"]) == 0
assert json.loads(output.getvalue())["schema_version"] == "v1"
assert errors.getvalue() == ""
def test_paper_read_supports_markdown_and_versioned_json(monkeypatch):
paths = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
paths.append((path, params))
return Response(b"# Paper\n\nBody", {})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "read", "2501.01234"]) == 0
assert output.getvalue() == "# Paper\n\nBody\n"
output = io.StringIO()
with redirect_stdout(output):
assert main(["paper", "read", "2501.01234", "--json"]) == 0
payload = json.loads(output.getvalue())
assert payload == {
"schema_version": "v1",
"data": {"paper": "2501.01234", "markdown": "# Paper\n\nBody"},
}
assert paths == [
("research/papers/2501.01234/read", None),
("research/papers/2501.01234/read", None),
]
def test_transport_is_anonymous_by_default(monkeypatch):
requests = []
class HTTPResponse(io.BytesIO):
headers = {}
def __enter__(self):
return self
def __exit__(self, *_args):
self.close()
def fake_urlopen(request, *, timeout):
requests.append(request)
assert timeout == 30
return HTTPResponse(b"# Public paper\n")
monkeypatch.delenv("PWC_API_URL", raising=False)
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
response = Client("https://example.test/api/v1").get(
"research/papers/2501.01234/read"
)
assert response.body == b"# Public paper\n"
assert requests[0].full_url.endswith("/api/v1/research/papers/2501.01234/read")
assert requests[0].get_header("Authorization") is None
def test_transport_timeout_is_configurable_without_changing_the_default(monkeypatch):
observed = []
class HTTPResponse(io.BytesIO):
headers = {}
def __enter__(self):
return self
def __exit__(self, *_args):
self.close()
def fake_urlopen(_request, *, timeout):
observed.append(timeout)
return HTTPResponse(b"{}")
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
Client("https://example.test/api/v1", timeout=25).get("health")
Client("https://example.test/api/v1").get("health")
assert observed == [25, 30]
def test_transport_encodes_repeated_query_parameters(monkeypatch):
requests = []
class HTTPResponse(io.BytesIO):
headers = {}
def __enter__(self):
return self
def __exit__(self, *_args):
self.close()
def fake_urlopen(request, *, timeout):
requests.append(request)
return HTTPResponse(b'{"count":0,"results":[]}')
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
Client("https://example.test/api/v1").get(
"papers/", {"author": ["Kaiming He", "@yilundu"]}
)
assert requests[0].full_url.endswith(
"/api/v1/papers/?author=Kaiming+He&author=%40yilundu"
)
def test_transport_reports_read_timeouts_and_socket_errors_as_transport_errors(
monkeypatch,
):
from pwc_cli.transport import TransportError
failures = iter([TimeoutError("The read operation timed out"), ConnectionResetError(104, "reset")])
def fake_urlopen(_request, *, timeout):
raise next(failures)
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
client = Client("https://example.test/api/v1")
with pytest.raises(TransportError, match="timed out"):
client.get("health")
with pytest.raises(TransportError, match="API request failed"):
client.get("health")
def test_closest_results_rank_by_shared_words_not_search_order():
from pwc_cli.cli import _closest_names, _exact_entity_match
candidates = [
{"name": "Spoken Language Identification"},
{"name": "Face Recognition"},
{"name": "Person Re-Identification (Video)"},
{"name": "Unsupervised Person Re-Identification"},
{"name": "Image Classification"},
]
assert _closest_names("person re-identification", candidates) == (
"Person Re-Identification (Video), Unsupervised Person Re-Identification"
)
with pytest.raises(ResponseError, match="closest results: Person Re-Identification"):
_exact_entity_match("person re-identification", candidates, label="Task")
def test_closest_results_omit_unrelated_candidates():
from pwc_cli.cli import _closest_names, _exact_entity_match
# Production chat: "FaceNet" suggested WaveNet, and "person
# re-identification" suggested Spoken Language Identification.
assert _closest_names("FaceNet", [{"name": "WaveNet"}, {"name": "ResNet"}]) == ""
assert (
_closest_names(
"person re-identification",
[{"name": "Spoken Language Identification"}, {"name": "Face Recognition"}],
)
== ""
)
with pytest.raises(ResponseError) as error:
_exact_entity_match("FaceNet", [{"name": "WaveNet"}], label="Method")
assert str(error.value) == "Method not found: FaceNet"
# Near spellings and containment still produce a hint.
assert _closest_names("Transfomer", [{"name": "Transformer"}]) == "Transformer"
assert _closest_names("ViT", [{"name": "ViTDet"}]) == ""
assert _closest_names("Swin", [{"name": "Swin Transformer"}]) == "Swin Transformer"
def test_metric_requests_resolve_through_case_and_common_aliases():
from pwc_cli.cli import _resolve_metric_name, _select_metric_rows
available = {"map": "mAP", "fps": "FPS", "top 1 accuracy": "Top 1 Accuracy"}
assert _resolve_metric_name("AP", available) == "mAP"
assert _resolve_metric_name("box ap", available) == "mAP"
assert _resolve_metric_name("top1", available) == "Top 1 Accuracy"
assert _resolve_metric_name("Top-1 Accuracy", available) == "Top 1 Accuracy"
assert _resolve_metric_name("fps", available) == "FPS"
assert _resolve_metric_name("latency", available) is None
rows = [
{"model_name": "Fast", "metrics": {"mAP": 58, "FPS": 100}},
{"model_name": "Slow", "metrics": {"mAP": 62, "FPS": 20}},
]
args = argparse.Namespace(
require_metrics=["ap"],
minimum_metrics=[("AP", 60)],
maximum_metrics=[],
sort_metric=("ap", "desc"),
pareto=[],
)
assert [row["model_name"] for row in _select_metric_rows(rows, args)] == ["Slow"]
assert args.sort_metric == ("mAP", "desc")
assert args.minimum_metrics == [("mAP", 60)]
unknown = argparse.Namespace(
require_metrics=[],
minimum_metrics=[],
maximum_metrics=[],
sort_metric=("latency", "desc"),
pareto=[],
)
with pytest.raises(UsageError, match="unknown metric\\(s\\): latency; available metrics: FPS, mAP"):
_select_metric_rows(rows, unknown)
def test_metric_requests_resolve_by_containment_and_alias_containment():
from pwc_cli.cli import _resolve_metric_name
# Requests seen from chat agents against production leaderboards.
assert (
_resolve_metric_name("D4RL Normalized Score", {"normalized score": "Normalized Score"})
== "Normalized Score"
)
assert _resolve_metric_name("Overall", {"geneval score": "GenEval Score"}) == "GenEval Score"
assert _resolve_metric_name("Score", {"dpg-bench score": "DPG-Bench Score"}) == "DPG-Bench Score"
assert _resolve_metric_name("Pass@1", {"accuracy": "Accuracy"}) == "Accuracy"
assert _resolve_metric_name("Pass Rate", {"success rate": "Success Rate"}) == "Success Rate"
mvtec = {
"detection auroc": "Detection AUROC",
"segmentation auroc": "Segmentation AUROC",
"detection ap": "Detection AP",
}
assert _resolve_metric_name("image-level AUROC", mvtec) == "Detection AUROC"
assert _resolve_metric_name("pixel-level AUROC", mvtec) == "Segmentation AUROC"
# Ambiguous or unrelated requests still surface the leaderboard's names.
assert _resolve_metric_name("AUROC", mvtec) is None
assert (
_resolve_metric_name(
"F-score",
{"mean f1 (advanced)": "Mean F1 (Advanced)", "mean f1 (intermediate)": "Mean F1 (Intermediate)"},
)
is None
)
assert _resolve_metric_name("latency", {"map": "mAP"}) is None
def test_entity_names_match_ignoring_punctuation_and_trailing_acronyms():
from pwc_cli.cli import _exact_entity_match, _task_match, _without_abbreviation
methods = [
{"name": "Mixture-of-Experts (MoE)", "slug": "moe"},
{"name": "Vision Transformer (ViT)", "slug": "vit"},
]
assert _exact_entity_match("mixture of experts", methods, label="Method")["slug"] == "moe"
assert _exact_entity_match("vision transformer", methods, label="Method")["slug"] == "vit"
with pytest.raises(ResponseError, match="Method not found: experts"):
_exact_entity_match("experts", methods, label="Method")
tasks = [
{"name": "Time Series Forecasting", "slug": "time-series-forecasting"},
{"name": "Person Re-Identification (Video)", "slug": "person-re-identification-video"},
{"name": "Unsupervised Person Re-Identification", "slug": "unsupervised-person-re-identification"},
]
assert _task_match("time-series forecasting", tasks)["slug"] == "time-series-forecasting"
# "(Video)" qualifies the task instead of abbreviating it, so the bare name
# must not silently pick the video variant.
assert _task_match("person re-identification", tasks) is None
# Two candidates that normalize alike stay unresolved.
assert (
_task_match(
"image-classification",
[{"name": "Image Classification"}, {"name": "Image, Classification"}],
)
is None
)
assert (
_without_abbreviation("Physics-Informed Neural Networks (PINNs)")
== "Physics-Informed Neural Networks"
)
assert _without_abbreviation("Aspect-Based Sentiment Analysis (ABSA)") == "Aspect-Based Sentiment Analysis"
assert _without_abbreviation("Person Re-Identification (Video)") is None
assert _without_abbreviation("MVTec AD (Unified Multi-Class)") is None
assert _without_abbreviation("Knowledge Distillation") is None
PUBLIC_METADATA = {
"in_catalog": False,
"catalog_id": None,
"catalog_url": None,
"metadata_source": "huggingface",
"note": "Not indexed on Papers With Code.",
"id": "2610.18869",
"arxiv_id": "2610.18869",
"doi": None,
"title": "A Paper Outside The Catalog",
"authors": ["Ada Lovelace", "Alan Turing"],
"abstract": "An abstract.",
"published": "2026-10-12",
"url": "https://arxiv.org/abs/2610.18869",
"pdf_url": "https://arxiv.org/pdf/2610.18869",
"hf_url": "https://huggingface.co/papers/2610.18869",
"repositories": [],
"project_pages": [],
"available_at": [
{"label": "arXiv", "url": "https://arxiv.org/abs/2610.18869", "kind": "abstract"},
{"label": "Hugging Face Papers", "url": "https://huggingface.co/papers/2610.18869", "kind": "huggingface"},
],
}
def _outside_catalog_client(public):
paths = []
class Client:
def __init__(self):
pass
def get(self, path, params=None):
paths.append(path)
if path.endswith("/public-metadata"):
if path in public:
return Response(json.dumps(public[path]).encode(), {})
raise HTTPStatusError(404, '{"detail":"Paper not found"}')
if path in {"papers/1706.03762", "research/papers/1706.03762/read"}:
if path.endswith("/read"):
return Response(b"# Attention\n", {})
return Response(
json.dumps({"id": "755", "title": "Attention Is All You Need"}).encode(),
{},
)
if path == "papers/search":
return Response(json.dumps({"results": []}).encode(), {})
raise HTTPStatusError(404, '{"detail":"Paper not found"}')
return Client, paths
@pytest.mark.parametrize("command", ["info", "read"])
@pytest.mark.parametrize("json_flag", [[], ["--json"]])
def test_paper_outside_the_catalog_shows_where_it_is_available(
monkeypatch, command, json_flag
):
Client, paths = _outside_catalog_client(
{"papers/2610.18869/public-metadata": PUBLIC_METADATA}
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output, errors = io.StringIO(), io.StringIO()
with redirect_stdout(output), redirect_stderr(errors):
# The exit status and stdout of a miss are unchanged for scripts.
assert main(["paper", command, "2610.18869", *json_flag]) == 3
assert output.getvalue() == ""
assert errors.getvalue() == (
"Not indexed on Papers With Code — available at:\n"
"- arXiv: https://arxiv.org/abs/2610.18869\n"
"- Hugging Face Papers: https://huggingface.co/papers/2610.18869\n"
"\n"
"title: A Paper Outside The Catalog\n"
"published: 2026-10-12\n"
"authors: Ada Lovelace, Alan Turing\n"
"metadata_source: huggingface\n"
"\n"
"## Abstract\n"
"\n"
"An abstract.\n"
"\n"
'pwc: API request failed (404): {"detail":"Paper not found"}\n'
)
assert paths[-1] == "papers/2610.18869/public-metadata"
def test_paper_title_outside_the_catalog_keeps_its_exit_status(monkeypatch):
path = "papers/Some%20Unindexed%20Title/public-metadata"
Client, _paths = _outside_catalog_client({path: PUBLIC_METADATA})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
errors = io.StringIO()
with redirect_stdout(io.StringIO()), redirect_stderr(errors):
assert main(["paper", "info", "Some Unindexed Title"]) == 4
assert errors.getvalue().startswith("Not indexed on Papers With Code — available at:\n")
assert errors.getvalue().endswith(
"pwc: Paper title not found: Some Unindexed Title\n"
)
def test_paper_unknown_everywhere_keeps_the_plain_error(monkeypatch):
Client, paths = _outside_catalog_client({})
monkeypatch.setattr("pwc_cli.cli.Client", Client)
errors = io.StringIO()
with redirect_stdout(io.StringIO()), redirect_stderr(errors):
assert main(["paper", "info", "2610.99999"]) == 3
assert errors.getvalue() == (
'pwc: API request failed (404): {"detail":"Paper not found"}\n'
)
assert paths == ["papers/2610.99999", "papers/2610.99999/public-metadata"]
@pytest.mark.parametrize(
("command", "expected"),
[("info", "title: Attention Is All You Need\n"), ("read", "# Attention\n")],
)
def test_paper_reference_the_api_finds_in_the_catalog_is_retried(
monkeypatch, command, expected
):
# The CLI resolves only IDs and titles; the API also maps arXiv URLs.
url = "https://arxiv.org/abs/1706.03762v7"
record = {**PUBLIC_METADATA, "in_catalog": True, "catalog_id": "755",
"arxiv_id": "1706.03762"}
Client, _paths = _outside_catalog_client(
{f"papers/{url}/public-metadata": record}
)
monkeypatch.setattr("pwc_cli.cli.Client", Client)
output, errors = io.StringIO(), io.StringIO()
with redirect_stdout(output), redirect_stderr(errors):
assert main(["paper", command, url]) == 0
assert expected in output.getvalue()
assert errors.getvalue() == ""
def test_embedded_paper_queries_do_not_call_the_public_metadata_api():
from pwc_cli.queries import query
Client, paths = _outside_catalog_client(
{"papers/2610.18869/public-metadata": PUBLIC_METADATA}
)
with pytest.raises(HTTPStatusError):
query(("paper", "info"), {"paper": "2610.18869"}, Client())
assert not any(path.endswith("/public-metadata") for path in paths)