Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ dynamic = ["version"]
[project.optional-dependencies]
build = ["uv ~= 0.8.17"]

[project.scripts]
usajobsapi = "usajobsapi.cli:main"

[tool.uv.sources]
mkdocs_external_files = { path = "mkdocs/mkdocs_external_files" }

Expand Down
229 changes: 225 additions & 4 deletions tests/unit/test_cli.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,227 @@
"""Placeholder tests for USAJobs API package."""
"""Unit tests for the CLI entry points."""

from __future__ import annotations

def test_placeholder() -> None:
"""A trivial test to ensure the test suite runs."""
assert True
import argparse
import json
import sys
from typing import Any, ClassVar

import pytest
from pydantic import BaseModel

from usajobsapi import cli


class DummyResponse(BaseModel):
"""Lightweight response model used to emulate client responses."""

payload: dict[str, Any]


class FakeClient:
"""Test double for ``USAJobsClient`` used to capture CLI interactions."""

instances: ClassVar[list["FakeClient"]] = []

def __init__(
self,
url: str | None = "https://data.usajobs.gov",
ssl_verify: bool = True,
timeout: float | None = 60,
auth_user: str | None = None,
auth_key: str | None = None,
session: Any = None,
) -> None:
self.url = url
self.ssl_verify = ssl_verify
self.timeout = timeout
self.auth_user = auth_user
self.auth_key = auth_key
self.session = session
self.headers = {
"Host": "data.usajobs.gov",
"User-Agent": auth_user,
"Authorization-Key": auth_key,
}
self.calls: list[tuple[str, dict[str, Any]]] = []
FakeClient.instances.append(self)

def _respond(self, method_name: str, params: dict[str, Any]) -> DummyResponse:
recorded = dict(params)
self.calls.append((method_name, recorded))
return DummyResponse(payload={"method": method_name, "params": recorded})

def announcement_text(self, **kwargs: Any) -> DummyResponse:
return self._respond("announcement_text", kwargs)

def search_jobs(self, **kwargs: Any) -> DummyResponse:
return self._respond("search_jobs", kwargs)

def historic_joa(self, **kwargs: Any) -> DummyResponse:
return self._respond("historic_joa", kwargs)


@pytest.fixture()
def fake_client(monkeypatch: pytest.MonkeyPatch) -> type[FakeClient]:
"""Patch ``USAJobsClient`` with the fake client for a single test."""
FakeClient.instances.clear()
monkeypatch.setattr(cli, "USAJobsClient", FakeClient)
return FakeClient


def test_parse_json_valid_object() -> None:
"""Ensure JSON parsing returns a dictionary."""
payload = cli._parse_json('{"foo": "bar"}')
assert payload == {"foo": "bar"}


@pytest.mark.parametrize(
"raw",
[
"{invalid",
'["not", "an", "object"]',
],
)
def test_parse_json_invalid_inputs(raw: str) -> None:
"""Invalid JSON strings raise argparse errors."""
with pytest.raises(argparse.ArgumentTypeError):
cli._parse_json(raw)


def test_build_parser_defaults() -> None:
"""The parser should expose the expected default values."""
parser = cli._build_parser()
args = parser.parse_args([cli.ACTIONS.TEXT.value])

assert args.action == cli.ACTIONS.TEXT
assert args.data == {}
assert args.prettify is False
assert args.ssl_verify is True
assert args.timeout == cli.USAJobsClient().timeout
assert args.auth_user is None
assert args.auth_key is None


def test_build_parser_custom_overrides() -> None:
"""Custom overrides should populate their respective fields."""
parser = cli._build_parser()
payload = {"keyword": "python"}
args = parser.parse_args(
[
cli.ACTIONS.SEARCH.value,
"-d",
json.dumps(payload),
"--prettify",
"--no-ssl-verify",
"--timeout",
"10",
"-A",
"user@example.com",
"--auth-key",
"secret",
]
)

assert args.action == cli.ACTIONS.SEARCH
assert args.data == payload
assert args.prettify is True
assert args.ssl_verify is False
assert args.timeout == 10.0
assert args.auth_user == "user@example.com"
assert args.auth_key == "secret"


def test_main_version_short_circuit(
monkeypatch: pytest.MonkeyPatch, capfd: pytest.CaptureFixture[str]
) -> None:
"""--version should print the package version and exit."""
monkeypatch.setattr(sys, "argv", ["prog", "--version"], raising=False)

with pytest.raises(SystemExit) as exc:
cli.main()

assert exc.value.code == 0
out = capfd.readouterr().out.strip()
assert out == cli.pkg_version


@pytest.mark.parametrize(
("action", "expected_method"),
[
(cli.ACTIONS.TEXT, "announcement_text"),
(cli.ACTIONS.SEARCH, "search_jobs"),
(cli.ACTIONS.HISTORIC, "historic_joa"),
],
)
def test_main_dispatches_to_expected_client_method(
monkeypatch: pytest.MonkeyPatch,
capfd: pytest.CaptureFixture[str],
fake_client: type[FakeClient],
action: cli.ACTIONS,
expected_method: str,
) -> None:
"""Each CLI action should invoke the corresponding client method."""
payload = {"query": action.value}
monkeypatch.setattr(
sys,
"argv",
["prog", action.value, "-d", json.dumps(payload)],
raising=False,
)

cli.main()

out = capfd.readouterr().out.strip()
assert json.loads(out) == {
"payload": {"method": expected_method, "params": payload}
}

instance = fake_client.instances[-1]
assert instance.calls[-1] == (expected_method, payload)


def test_main_prettify_outputs_indented_json(
monkeypatch: pytest.MonkeyPatch,
capfd: pytest.CaptureFixture[str],
fake_client: type[FakeClient],
) -> None:
"""--prettify should request indented JSON output."""
payload = {"foo": "bar"}
monkeypatch.setattr(
sys,
"argv",
["prog", cli.ACTIONS.SEARCH.value, "-d", json.dumps(payload), "--prettify"],
raising=False,
)

cli.main()
out = capfd.readouterr().out

assert out.startswith("{\n")
assert '"payload"' in out

instance = fake_client.instances[-1]
assert instance.calls[-1] == ("search_jobs", payload)


def test_main_defaults_to_empty_payload(
monkeypatch: pytest.MonkeyPatch,
capfd: pytest.CaptureFixture[str],
fake_client: type[FakeClient],
) -> None:
"""When no JSON payload is supplied the client receives an empty dict."""
monkeypatch.setattr(
sys,
"argv",
["prog", cli.ACTIONS.SEARCH.value],
raising=False,
)

cli.main()

out = capfd.readouterr().out.strip()
assert json.loads(out) == {"payload": {"method": "search_jobs", "params": {}}}

instance = fake_client.instances[-1]
assert instance.calls[-1] == ("search_jobs", {})
143 changes: 142 additions & 1 deletion usajobsapi/cli.py
Original file line number Diff line number Diff line change
@@ -1,2 +1,143 @@
"""
Command line interface for the python-usajobsapi package.

This module powers the `usajobsapi` executable, so that a user can query the exposed endpoints from the command line.
"""

from __future__ import annotations

import argparse
import json
import sys
from enum import StrEnum
from typing import Any

from pydantic import BaseModel

from usajobsapi._version import __title__ as pkg_title
from usajobsapi._version import __version__ as pkg_version
from usajobsapi.client import USAJobsClient


class ACTIONS(StrEnum):
TEXT = "announcementtext"
SEARCH = "search"
HISTORIC = "historicjoa"


def _parse_json(value: str) -> dict[str, Any]:
"""Parse JSON-encoded argument data."""
try:
parsed = json.loads(value)
except json.JSONDecodeError as exc:
raise argparse.ArgumentTypeError(f"Invalid JSON payload: {exc}") from exc

if not isinstance(parsed, dict):
raise argparse.ArgumentTypeError(
"JSON payload must decode to an object (dict)."
)

return parsed


def _build_parser() -> argparse.ArgumentParser:
"""Create the top-level argument parser for the CLI."""
client_defaults = USAJobsClient()
parser = argparse.ArgumentParser(
prog=pkg_title,
description="USAJOBS REST API Command Line Interface.",
)

parser.add_argument(
"--version",
action="version",
version=f"%(prog)s {pkg_version}",
help="Display the package version.",
)

parser.add_argument(
"action",
type=ACTIONS,
choices=list(ACTIONS),
help="Endpoint to query.",
)
parser.add_argument(
"-d",
"--data",
type=_parse_json,
default={},
metavar="JSON",
help="JSON-encoded parameters to pass to the selected endpoint.",
)
parser.add_argument(
"--prettify", action="store_true", help="Prettify the JSON output."
)

parser.add_argument(
"--no-ssl-verify",
dest="ssl_verify",
action="store_false",
default=client_defaults.ssl_verify,
help="Disable TLS certificate verification.",
)
parser.add_argument(
"--timeout",
type=float,
default=client_defaults.timeout,
metavar="SECONDS",
help="Request timeout in seconds. Defaults to the client default.",
)

parser.add_argument(
"-A",
"--user-agent",
dest="auth_user",
default=client_defaults.headers["User-Agent"],
metavar="EMAIL",
help="Email address associated with the API key (User-Agent header).",
)
parser.add_argument(
"--auth-key",
dest="auth_key",
default=client_defaults.headers["Authorization-Key"],
metavar="KEY",
help="API key used for authenticated requests.",
)

return parser


def main() -> None:
pass
if "--version" in sys.argv:
print(pkg_version)
sys.exit(0)

parser = _build_parser()
args = parser.parse_args()

client = USAJobsClient(
ssl_verify=args.ssl_verify,
timeout=args.timeout,
auth_user=args.auth_user,
auth_key=args.auth_key,
)

action = args.action
if not action:
sys.exit(0)

resp: BaseModel | None = None
if action == ACTIONS.TEXT:
resp = client.announcement_text(**args.data)
elif action == ACTIONS.SEARCH:
resp = client.search_jobs(**args.data)
elif action == ACTIONS.HISTORIC:
resp = client.historic_joa(**args.data)

if not resp:
sys.exit(1)

if args.prettify:
print(resp.model_dump_json(indent=2))
else:
print(resp.model_dump_json())