Add mock API for testing purposes

pyproject.toml

@@ -24,6 +24,8 @@ isort = "^5.8.0"
 mypy = "^0.812"
 pytest = "^6.2.3"
 python-dotenv = "^0.17.0"
+fastapi = "^0.64.0"
+uvicorn = {extras = ["standard"], version = "^0.13.4"}
 
 [tool.poetry.scripts]
 ipums = "ipumspy.cli:ipums_group"

src/ipumspy/api/core.py

@@ -292,10 +292,12 @@ def retrieve_previous_extracts(
             collections = [collection]
 
         # TODO: Wrap results in Extract objects.
-        return {
-            collection: self.get(
-                self.base_url,
-                params={"collection": collection, "limit": limit, "version": "v1"},
-            ).json()
-            for collection in collections
-        }
+        output = {}
+        for collection in collections:
+            output.update(
+                self.get(
+                    self.base_url,
+                    params={"collection": collection, "limit": limit, "version": "v1"},
+                ).json()
+            )
+        return output

src/ipumspy/api/extract.py

@@ -23,6 +23,7 @@ def __init__(self):
     def __init_subclass__(cls, collection: str, **kwargs):
         super().__init_subclass__(**kwargs)
         cls.collection = collection
+        BaseExtract._collection_to_extract[collection] = cls
 
     def build(self) -> Dict[str, Any]:
         """

src/ipumspy/cli.py

@@ -347,4 +347,4 @@ def convert_command(
 
 
 if __name__ == "__main__":
-    ipums_group()
+    ipums_group()

tests/mock_api.py

@@ -0,0 +1,37 @@
+import os
+from typing import Dict, Optional
+
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from pydantic import BaseModel
+
+load_dotenv(".env.test")
+
+app = FastAPI()
+
+
+class ExtractSpec(BaseModel):
+    description: str
+    data_format: str
+    data_structure: Dict[str, dict]
+    samples: Dict[str, dict]
+    variables: Dict[str, dict]
+
+
+@app.post("/extracts")
+async def submit_extract(
+    collection: str, extract: ExtractSpec, request: Request, version: str = "v1",
+):
+    if request.headers["Authorization"] != os.environ.get("IPUMS_API_KEY"):
+        raise HTTPException(403, "Incorrect api key")
+    return {"number": 10}
+
+
+@app.get("/extracts")
+async def retrieve_previous_extracts(
+    collection: str, request: Request, limit: int = 10, version: str = "v1",
+):
+    if request.headers["Authorization"] != os.environ.get("IPUMS_API_KEY"):
+        raise HTTPException(403, "Incorrect api key")
+
+    return {collection: list(range(10))}

tests/test_api.py

@@ -1,13 +1,30 @@
 import os
+import subprocess
+import time
 
 import pytest
 
 from ipumspy.api import CpsExtract, IpumsApiClient, OtherExtract
 
 
+@pytest.fixture(scope="module")
+def mock_api() -> str:
+    # TODO: Would be good to randomly assign a port and return it
+    p = subprocess.Popen(
+        ["uvicorn", "tests.mock_api:app", "--host", "127.0.0.1", "--port", "8989"]
+    )
+    time.sleep(1)  # Give it enough time to warm up
+    try:
+        yield "http://127.0.0.1:8989/extracts"
+    finally:
+        p.kill()
+
+
 @pytest.fixture(scope="function")
-def api_client(environment_variables) -> IpumsApiClient:
-    return IpumsApiClient(os.environ.get("IPUMS_API_KEY"))
+def api_client(environment_variables, mock_api: str) -> IpumsApiClient:
+    client = IpumsApiClient(os.environ.get("IPUMS_API_KEY"))
+    client.base_url = mock_api
+    return client
 
 
 def test_cps_build_extract():
@@ -17,6 +34,7 @@ def test_cps_build_extract():
     extract = CpsExtract(
         ["cps1976_01s", "cps1976_02b"], ["YEAR", "MISH", "AGE", "RACE", "UH_SEX_B1"],
     )
+    assert extract.collection == "cps"
     assert extract.build() == {
         "data_structure": {"rectangular": {"on": "P"}},
         "samples": {"cps1976_01s": {}, "cps1976_02b": {}},
@@ -42,10 +60,14 @@ def test_submit_extract(api_client: IpumsApiClient):
     )
 
     api_client.submit_extract(extract)
-    assert extract.extract_id is not None
+    assert extract.extract_id == 10
 
 
 def test_retrieve_previous_extracts(api_client: IpumsApiClient):
+    previous = api_client.retrieve_previous_extracts(collection="cps")
+    assert "cps" in previous
+    assert len(previous["cps"]) == 10
+
     previous = api_client.retrieve_previous_extracts()
     assert "cps" in previous
     assert len(previous["cps"]) == 10