feat(api): support the historic joa api endpoint

tests/conftest.py

@@ -0,0 +1,139 @@
+from typing import Dict, List
+
+import pytest
+
+
+@pytest.fixture
+def historicjoa_params_kwargs() -> Dict[str, str]:
+    """Field-value mapping used to build HistoricJoaEndpoint params models."""
+
+    return {
+        "hiring_agency_codes": "AGENCY1",
+        "hiring_department_codes": "DEPT1",
+        "position_series": "2210",
+        "announcement_numbers": "23-ABC",
+        "usajobs_control_numbers": "1234567",
+        "start_position_open_date": "2020-01-01",
+        "end_position_open_date": "2020-12-31",
+        "start_position_close_date": "2021-01-01",
+        "end_position_close_date": "2021-12-31",
+        "continuation_token": "token123",
+    }
+
+
+@pytest.fixture
+def historicjoa_response_payload() -> Dict[str, object]:
+    """Serialized Historic JOA response payload mimicking the USAJOBS API."""
+
+    return {
+        "paging": {
+            "metadata": {
+                "totalCount": 2,
+                "pageSize": 2,
+                "continuationToken": "NEXTTOKEN",
+            },
+            "next": "https://example.invalid/historicjoa?page=2",
+        },
+        "data": _historicjoa_items(),
+    }
+
+
+def _historicjoa_items() -> List[Dict[str, object]]:
+    return [
+        {
+            "usajobsControlNumber": 123456789,
+            "hiringAgencyCode": "NASA",
+            "hiringAgencyName": "National Aeronautics and Space Administration",
+            "hiringDepartmentCode": "NAT",
+            "hiringDepartmentName": "Department of Science",
+            "agencyLevel": 2,
+            "agencyLevelSort": "Department of Science\\NASA",
+            "appointmentType": "Permanent",
+            "workSchedule": "Full-time",
+            "payScale": "GS",
+            "salaryType": "Per Year",
+            "vendor": "USASTAFFING",
+            "travelRequirement": "Occasional travel",
+            "teleworkEligible": "Y",
+            "serviceType": "Competitive",
+            "securityClearanceRequired": "Y",
+            "securityClearance": "Secret",
+            "whoMayApply": "United States Citizens",
+            "announcementClosingTypeCode": "C",
+            "announcementClosingTypeDescription": "Closing Date",
+            "positionOpenDate": "2020-01-01",
+            "positionCloseDate": "2020-02-01",
+            "positionExpireDate": None,
+            "announcementNumber": "NASA-20-001",
+            "hiringSubelementName": "Space Operations",
+            "positionTitle": "Data Scientist",
+            "minimumGrade": "12",
+            "maximumGrade": "13",
+            "promotionPotential": "13",
+            "minimumSalary": 90000.0,
+            "maximumSalary": 120000.0,
+            "supervisoryStatus": "N",
+            "drugTestRequired": "N",
+            "relocationExpensesReimbursed": "Y",
+            "totalOpenings": "3",
+            "disableApplyOnline": "N",
+            "positionOpeningStatus": "Accepting Applications",
+            "hiringPaths": [{"hiringPath": "The public"}],
+            "jobCategories": [{"series": "1550"}],
+            "positionLocations": [
+                {
+                    "positionLocationCity": "Houston",
+                    "positionLocationState": "Texas",
+                    "positionLocationCountry": "United States",
+                }
+            ],
+        },
+        {
+            "usajobsControlNumber": 987654321,
+            "hiringAgencyCode": "DOE",
+            "hiringAgencyName": "Department of Energy",
+            "hiringDepartmentCode": "ENG",
+            "hiringDepartmentName": "Department of Energy",
+            "agencyLevel": 1,
+            "agencyLevelSort": "Department of Energy",
+            "appointmentType": "Term",
+            "workSchedule": "Part-time",
+            "payScale": "GS",
+            "salaryType": "Per Year",
+            "vendor": "OTHER",
+            "travelRequirement": "Not required",
+            "teleworkEligible": "N",
+            "serviceType": None,
+            "securityClearanceRequired": "N",
+            "securityClearance": "Not Required",
+            "whoMayApply": "Agency Employees Only",
+            "announcementClosingTypeCode": None,
+            "announcementClosingTypeDescription": None,
+            "positionOpenDate": "2020-03-01",
+            "positionCloseDate": "2020-04-01",
+            "positionExpireDate": "2020-04-15",
+            "announcementNumber": "DOE-20-ENG",
+            "hiringSubelementName": "Energy Research",
+            "positionTitle": "Backend Engineer",
+            "minimumGrade": "11",
+            "maximumGrade": "12",
+            "promotionPotential": None,
+            "minimumSalary": 80000.0,
+            "maximumSalary": 110000.0,
+            "supervisoryStatus": "Y",
+            "drugTestRequired": "Y",
+            "relocationExpensesReimbursed": "N",
+            "totalOpenings": "1",
+            "disableApplyOnline": "Y",
+            "positionOpeningStatus": "Closed",
+            "hiringPaths": [{"hiringPath": "Government employees"}],
+            "jobCategories": [{"series": "2210"}],
+            "positionLocations": [
+                {
+                    "positionLocationCity": "Washington",
+                    "positionLocationState": "District of Columbia",
+                    "positionLocationCountry": "United States",
+                }
+            ],
+        },
+    ]

tests/unit/test_client.py

@@ -1,6 +1,197 @@
-"""Placeholder tests for USAJobs API package."""
+"""Unit tests for USAJobsApiClient."""
 
+from copy import deepcopy
 
-def test_placeholder() -> None:
-    """A trivial test to ensure the test suite runs."""
-    assert True
+import pytest
+
+from usajobsapi.client import USAJobsApiClient
+from usajobsapi.endpoints.historicjoa import HistoricJoaEndpoint
+
+# test historic_joa_pages
+# ---
+
+
+def test_historic_joa_pages_yields_pages(
+    monkeypatch, historicjoa_response_payload
+) -> None:
+    """Ensure historic_joa_pages yields pages while forwarding continuation tokens."""
+
+    first_payload = deepcopy(historicjoa_response_payload)
+    second_payload = deepcopy(historicjoa_response_payload)
+    second_payload["paging"]["metadata"]["continuationToken"] = None
+    second_payload["data"] = []
+
+    responses = [
+        HistoricJoaEndpoint.Response.model_validate(first_payload),
+        HistoricJoaEndpoint.Response.model_validate(second_payload),
+    ]
+    captured_kwargs = []
+
+    def fake_historic_joa(self, **call_kwargs):
+        captured_kwargs.append(call_kwargs)
+        return responses.pop(0)
+
+    monkeypatch.setattr(USAJobsApiClient, "historic_joa", fake_historic_joa)
+
+    client = USAJobsApiClient()
+
+    pages = list(
+        client.historic_joa_pages(
+            hiring_agency_codes="NASA", continuation_token="INITIALTOKEN"
+        )
+    )
+
+    assert len(pages) == 2
+    assert pages[0].next_token() == "NEXTTOKEN"
+    assert pages[1].next_token() is None
+    assert captured_kwargs == [
+        {"hiring_agency_codes": "NASA", "continuation_token": "INITIALTOKEN"},
+        {"hiring_agency_codes": "NASA", "continuation_token": "NEXTTOKEN"},
+    ]
+
+
+def test_historic_joa_pages_duplicate_token(
+    monkeypatch, historicjoa_response_payload
+) -> None:
+    """Duplicate continuation tokens should raise to avoid infinite loops."""
+
+    first_response = HistoricJoaEndpoint.Response.model_validate(
+        historicjoa_response_payload
+    )
+    duplicate_payload = deepcopy(historicjoa_response_payload)
+    duplicate_payload["paging"]["metadata"]["continuationToken"] = (
+        first_response.next_token()
+    )
+    responses = [
+        first_response,
+        HistoricJoaEndpoint.Response.model_validate(duplicate_payload),
+    ]
+
+    def fake_historic_joa(self, **_):
+        return responses.pop(0)
+
+    monkeypatch.setattr(USAJobsApiClient, "historic_joa", fake_historic_joa)
+
+    client = USAJobsApiClient()
+    iterator = client.historic_joa_pages()
+
+    assert next(iterator)
+    with pytest.raises(RuntimeError, match="duplicate continuation token"):
+        next(iterator)
+
+
+# test historic_joa_items
+# ---
+
+
+def test_historic_joa_items_yields_items_across_pages(
+    monkeypatch: pytest.MonkeyPatch, historicjoa_response_payload
+) -> None:
+    """Ensure historic_joa_items yields items and follows continuation tokens."""
+
+    client = USAJobsApiClient()
+
+    first_page = deepcopy(historicjoa_response_payload)
+    first_page["paging"]["metadata"]["continuationToken"] = "TOKEN2"
+    first_page["data"] = first_page["data"][:2]
+
+    second_page = {
+        "paging": {
+            "metadata": {"totalCount": 3, "pageSize": 1, "continuationToken": None},
+            "next": None,
+        },
+        "data": [
+            {
+                "usajobsControlNumber": 111222333,
+                "hiringAgencyCode": "GSA",
+                "hiringAgencyName": "General Services Administration",
+                "hiringDepartmentCode": "GSA",
+                "hiringDepartmentName": "General Services Administration",
+                "agencyLevel": 1,
+                "agencyLevelSort": "GSA",
+                "appointmentType": "Permanent",
+                "workSchedule": "Full-time",
+                "payScale": "GS",
+                "salaryType": "Per Year",
+                "vendor": "USASTAFFING",
+                "travelRequirement": "Not required",
+                "teleworkEligible": "Y",
+                "serviceType": "Competitive",
+                "securityClearanceRequired": "N",
+                "securityClearance": "Not Required",
+                "whoMayApply": "All",
+                "announcementClosingTypeCode": "C",
+                "announcementClosingTypeDescription": "Closing Date",
+                "positionOpenDate": "2020-05-01",
+                "positionCloseDate": "2020-05-15",
+                "positionExpireDate": None,
+                "announcementNumber": "GSA-20-001",
+                "hiringSubelementName": "Administration",
+                "positionTitle": "Systems Analyst",
+                "minimumGrade": "11",
+                "maximumGrade": "12",
+                "promotionPotential": "13",
+                "minimumSalary": 85000.0,
+                "maximumSalary": 95000.0,
+                "supervisoryStatus": "N",
+                "drugTestRequired": "N",
+                "relocationExpensesReimbursed": "N",
+                "totalOpenings": "2",
+                "disableApplyOnline": "N",
+                "positionOpeningStatus": "Accepting Applications",
+                "hiringPaths": [{"hiringPath": "The public"}],
+                "jobCategories": [{"series": "2210"}],
+                "positionLocations": [
+                    {
+                        "positionLocationCity": "Washington",
+                        "positionLocationState": "District of Columbia",
+                        "positionLocationCountry": "United States",
+                    }
+                ],
+            }
+        ],
+    }
+
+    responses = [first_page, second_page]
+    calls = []
+
+    def fake_historic(**call_kwargs):
+        calls.append(call_kwargs)
+        return HistoricJoaEndpoint.Response.model_validate(responses.pop(0))
+
+    monkeypatch.setattr(client, "historic_joa", fake_historic)
+
+    items = list(client.historic_joa_items(hiring_agency_codes="NASA"))
+
+    assert [item.usajobs_control_number for item in items] == [
+        123456789,
+        987654321,
+        111222333,
+    ]
+    assert len(calls) == 2
+    assert "continuation_token" not in calls[0]
+    assert calls[1]["continuation_token"] == "TOKEN2"
+
+
+def test_historic_joa_items_respects_initial_token(
+    monkeypatch: pytest.MonkeyPatch, historicjoa_response_payload
+) -> None:
+    """Ensure historic_joa_items uses the supplied initial continuation token."""
+
+    client = USAJobsApiClient()
+
+    payload = deepcopy(historicjoa_response_payload)
+    payload["paging"]["metadata"]["continuationToken"] = None
+
+    calls = []
+
+    def fake_historic(**call_kwargs):
+        calls.append(call_kwargs)
+        return HistoricJoaEndpoint.Response.model_validate(payload)
+
+    monkeypatch.setattr(client, "historic_joa", fake_historic)
+
+    items = list(client.historic_joa_items(continuation_token="SEED"))
+
+    assert len(items) == len(payload["data"])
+    assert calls[0]["continuation_token"] == "SEED"