diff --git a/ogc_api_processes_fastapi/clients.py b/ogc_api_processes_fastapi/clients.py index 4a3ad87..eeb50ab 100644 --- a/ogc_api_processes_fastapi/clients.py +++ b/ogc_api_processes_fastapi/clients.py @@ -19,7 +19,7 @@ import fastapi -from . import responses +from . import models class BaseClient(abc.ABC): @@ -28,7 +28,7 @@ class BaseClient(abc.ABC): @abc.abstractmethod def get_processes( self, limit: Optional[int] = fastapi.Query(None) - ) -> responses.ProcessList: + ) -> models.ProcessList: """Get all available processes. Called with `GET /processes`. @@ -40,7 +40,7 @@ def get_processes( Returns ------- - responses.ProcessList + models.ProcessList List of available processes summaries. """ ... @@ -48,7 +48,7 @@ def get_processes( @abc.abstractmethod def get_process( self, process_id: str = fastapi.Path(...) - ) -> responses.ProcessDescription: + ) -> models.ProcessDescription: """Get description of the process identified by `process_id`. Called with `GET /processes/{process_id}`. @@ -60,7 +60,7 @@ def get_process( Returns ------- - responses.schema["ProcessDescription"] + models.ProcessDescription Description of the process. Raises @@ -75,7 +75,7 @@ def post_process_execution( self, process_id: str = fastapi.Path(...), execution_content: Dict[str, Any] = fastapi.Body(...), - ) -> responses.StatusInfo: + ) -> models.StatusInfo: """Post request for execution of the process identified by `process_id`. Called with `POST /processes/{process_id}/execution`. @@ -90,7 +90,7 @@ def post_process_execution( Returns ------- - responses.StatusInfo + models.StatusInfo Information on the status of the submitted job. Raises @@ -106,7 +106,7 @@ def get_jobs( processID: Optional[List[str]] = fastapi.Query(None), status: Optional[List[str]] = fastapi.Query(None), limit: Optional[int] = fastapi.Query(10, ge=1, le=10000), - ) -> responses.JobList: + ) -> models.JobList: """Get the list of submitted jobs. Called with `GET /jobs`. @@ -127,12 +127,12 @@ def get_jobs( Returns ------- - responses.JobList + models.JobList List of jobs. """ @abc.abstractmethod - def get_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: + def get_job(self, job_id: str = fastapi.Path(...)) -> models.StatusInfo: """Get status information of the job identified by `job_id`. Called with `GET /jobs/{job_id}`. @@ -144,7 +144,7 @@ def get_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: Returns ------- - responses.StatusInfo + models.StatusInfo Information on the status of the job. Raises @@ -155,7 +155,7 @@ def get_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: ... @abc.abstractmethod - def get_job_results(self, job_id: str = fastapi.Path(...)) -> responses.Results: + def get_job_results(self, job_id: str = fastapi.Path(...)) -> models.Results: """Get results of the job identified by `job_id`. Called with `GET /jobs/{job_id}/results`. @@ -167,7 +167,7 @@ def get_job_results(self, job_id: str = fastapi.Path(...)) -> responses.Results: Returns ------- - responses.Results + models.Results Job results. Raises @@ -182,7 +182,7 @@ def get_job_results(self, job_id: str = fastapi.Path(...)) -> responses.Results: ... @abc.abstractmethod - def delete_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: + def delete_job(self, job_id: str = fastapi.Path(...)) -> models.StatusInfo: """Cancel the job identified by `job_id`. Called with `DELETE /jobs/{job_id}`. @@ -194,7 +194,7 @@ def delete_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: Returns ------- - responses.StatusInfo + models.StatusInfo Information on the status of the job. Raises diff --git a/ogc_api_processes_fastapi/endpoints.py b/ogc_api_processes_fastapi/endpoints.py index ba8b477..3866e23 100644 --- a/ogc_api_processes_fastapi/endpoints.py +++ b/ogc_api_processes_fastapi/endpoints.py @@ -5,22 +5,22 @@ import fastapi -from . import clients, responses +from . import clients, models def create_links_to_job( - request: fastapi.Request, job: responses.StatusInfo -) -> List[responses.Link]: + request: fastapi.Request, job: models.StatusInfo +) -> List[models.Link]: """Create links to attach to provided the job. Parameters ---------- - job : schema["StatusInfo"] + job : models.StatusInfo Job to create links for. Returns ------- - List[responses.Link] + List[models.Link] Links to attach to job. """ rel_job_link = "self" @@ -29,7 +29,7 @@ def create_links_to_job( rel_job_link = "monitor" title_job_link = "job status info" links = [ - responses.Link( + models.Link( href=urllib.parse.urljoin(str(request.base_url), f"jobs/{job.jobID}"), rel=rel_job_link, type="application/json", @@ -38,7 +38,7 @@ def create_links_to_job( ] if job.status.value in ("successful", "failed"): links.append( - responses.Link( + models.Link( href=urllib.parse.urljoin( str(request.base_url), f"jobs/{job.jobID}/results" ), @@ -50,8 +50,8 @@ def create_links_to_job( def create_self_link( request_url: str, title: Optional[str] = None, type: Optional[str] = None -) -> responses.Link: - self_link = responses.Link(href=str(request_url), rel="self") +) -> models.Link: + self_link = models.Link(href=str(request_url), rel="self") if type: self_link.type = type if title: @@ -60,8 +60,8 @@ def create_self_link( def create_page_link( - request_url: str, page: str, pagination_qs: responses.PaginationQueryParameters -) -> responses.Link: + request_url: str, page: str, pagination_qs: models.PaginationQueryParameters +) -> models.Link: if page not in ("next", "prev"): raise ValueError(f"{page} is not a valid value for ``page`` parameter") request_parsed = urllib.parse.urlsplit(request_url) @@ -72,13 +72,13 @@ def create_page_link( query_string = urllib.parse.urlencode(queries_page, doseq=True) parsed_page_request = request_parsed._replace(query=query_string) url_page = parsed_page_request.geturl() - link_page = responses.Link(href=url_page, rel=page) + link_page = models.Link(href=url_page, rel=page) return link_page def create_pagination_links( - request_url: str, pagination_qs: Optional[responses.PaginationQueryParameters] -) -> List[responses.Link]: + request_url: str, pagination_qs: Optional[models.PaginationQueryParameters] +) -> List[models.Link]: pagination_links = [] if pagination_qs: if pagination_qs.next: @@ -92,32 +92,32 @@ def create_pagination_links( def create_get_landing_page_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request], responses.LandingPage]: +) -> Callable[[fastapi.Request], models.LandingPage]: def get_landing_page( request: fastapi.Request, - ) -> responses.LandingPage: + ) -> models.LandingPage: """Get the API landing page.""" links = [ - responses.Link( + models.Link( href=urllib.parse.urljoin(str(request.base_url), "openapi.json"), rel="service-desc", type="application/vnd.oai.openapi+json;version=3.0", title="OpenAPI service description", ), - responses.Link( + models.Link( href=urllib.parse.urljoin(str(request.base_url), "conformance"), rel="http://www.opengis.net/def/rel/ogc/1.0/conformance", type="application/json", title="Conformance declaration", ), - responses.Link( + models.Link( href=urllib.parse.urljoin(str(request.base_url), "processes"), rel="http://www.opengis.net/def/rel/ogc/1.0/processes", type="application/json", title="Metadata about the processes", ), ] - landing_page = responses.LandingPage(links=links) + landing_page = models.LandingPage(links=links) return landing_page @@ -126,10 +126,10 @@ def get_landing_page( def create_get_conformance_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request], responses.ConfClass]: - def get_conformance(request: fastapi.Request) -> responses.ConfClass: +) -> Callable[[fastapi.Request], models.ConfClass]: + def get_conformance(request: fastapi.Request) -> models.ConfClass: """Get the API conformance declaration page.""" - conformance = responses.ConfClass( + conformance = models.ConfClass( conformsTo=[ "http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/core", "http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/ogc-process-description", @@ -146,11 +146,11 @@ def get_conformance(request: fastapi.Request) -> responses.ConfClass: def create_get_processes_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request], responses.ProcessList]: +) -> Callable[[fastapi.Request], models.ProcessList]: def get_processes( request: fastapi.Request, - process_list: responses.ProcessList = fastapi.Depends(client.get_processes), - ) -> responses.ProcessList: + process_list: models.ProcessList = fastapi.Depends(client.get_processes), + ) -> models.ProcessList: """Get the list of available processes. The list of processes contains a summary of each process @@ -159,7 +159,7 @@ def get_processes( """ for process in process_list.processes: process.links = [ - responses.Link( + models.Link( href=urllib.parse.urljoin( str(request.base_url), f"processes/{process.id}" ), @@ -184,11 +184,11 @@ def get_processes( def create_get_process_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request], responses.ProcessDescription]: +) -> Callable[[fastapi.Request], models.ProcessDescription]: def get_process( request: fastapi.Request, - process: responses.ProcessDescription = fastapi.Depends(client.get_process), - ) -> responses.ProcessDescription: + process: models.ProcessDescription = fastapi.Depends(client.get_process), + ) -> models.ProcessDescription: """Get the description of a specific process. The list of processes contains a summary of each process @@ -197,7 +197,7 @@ def get_process( """ process.links = [ create_self_link(str(request.url)), - responses.Link( + models.Link( href=urllib.parse.urljoin( str(request.base_url), f"processes/{process.id}/execution" ), @@ -214,18 +214,16 @@ def get_process( def create_post_process_execution_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request, fastapi.Response], responses.StatusInfo]: +) -> Callable[[fastapi.Request, fastapi.Response], models.StatusInfo]: def post_process_execution( request: fastapi.Request, response: fastapi.Response, - status_info: responses.StatusInfo = fastapi.Depends( - client.post_process_execution - ), - ) -> responses.StatusInfo: + status_info: models.StatusInfo = fastapi.Depends(client.post_process_execution), + ) -> models.StatusInfo: """Create a new job.""" status_info.links = [ create_self_link(str(request.url)), - responses.Link( + models.Link( href=urllib.parse.urljoin( str(request.base_url), f"jobs/{status_info.jobID}" ), @@ -245,11 +243,11 @@ def post_process_execution( def create_get_jobs_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request], responses.JobList]: +) -> Callable[[fastapi.Request], models.JobList]: def get_jobs( request: fastapi.Request, - job_list: responses.JobList = fastapi.Depends(client.get_jobs), - ) -> responses.JobList: + job_list: models.JobList = fastapi.Depends(client.get_jobs), + ) -> models.JobList: """Show the list of submitted jobs.""" for job in job_list.jobs: job.links = create_links_to_job(job=job, request=request) @@ -269,11 +267,11 @@ def get_jobs( def create_get_job_endpoint( client: clients.BaseClient, -) -> Callable[[fastapi.Request], responses.StatusInfo]: +) -> Callable[[fastapi.Request], models.StatusInfo]: def get_job( request: fastapi.Request, - job: responses.StatusInfo = fastapi.Depends(client.get_job), - ) -> responses.StatusInfo: + job: models.StatusInfo = fastapi.Depends(client.get_job), + ) -> models.StatusInfo: """Show the status of a job.""" job.links = create_links_to_job(job=job, request=request) @@ -284,10 +282,10 @@ def get_job( def create_get_job_results_endpoint( client: clients.BaseClient, -) -> Callable[[], responses.Results]: +) -> Callable[[], models.Results]: def get_job_results( - job_results: responses.Results = fastapi.Depends(client.get_job_results), - ) -> responses.Results: + job_results: models.Results = fastapi.Depends(client.get_job_results), + ) -> models.Results: """Show results of a job.""" return job_results @@ -296,10 +294,10 @@ def get_job_results( def create_delete_job_endpoint( client: clients.BaseClient, -) -> Callable[[], responses.StatusInfo]: +) -> Callable[[], models.StatusInfo]: def delete_job( - job: responses.StatusInfo = fastapi.Depends(client.delete_job), - ) -> responses.StatusInfo: + job: models.StatusInfo = fastapi.Depends(client.delete_job), + ) -> models.StatusInfo: """Cancel a job.""" return job diff --git a/ogc_api_processes_fastapi/main.py b/ogc_api_processes_fastapi/main.py index 4e87b00..9150135 100644 --- a/ogc_api_processes_fastapi/main.py +++ b/ogc_api_processes_fastapi/main.py @@ -5,16 +5,16 @@ import fastapi import pydantic -from . import clients, config, endpoints, responses +from . import clients, config, endpoints, models def set_response_model( client: clients.BaseClient, route_name: str ) -> pydantic.BaseModel: if route_name == "GetLandingPage": - base_model = responses.LandingPage + base_model = models.LandingPage elif route_name == "GetConformance": - base_model = responses.ConfClass # type: ignore + base_model = models.ConfClass # type: ignore else: base_model = typing.get_type_hints( getattr(client, config.ROUTES[route_name].client_method) # type: ignore diff --git a/ogc_api_processes_fastapi/responses.py b/ogc_api_processes_fastapi/models.py similarity index 100% rename from ogc_api_processes_fastapi/responses.py rename to ogc_api_processes_fastapi/models.py diff --git a/tests/conftest.py b/tests/conftest.py index 66eff22..269a130 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,7 +17,7 @@ import fastapi import pytest -from ogc_api_processes_fastapi import clients, responses +from ogc_api_processes_fastapi import clients, models PROCESSES_DB = [ { @@ -37,25 +37,25 @@ class TestClientDefault(clients.BaseClient): def get_processes( self, limit: Optional[int] = fastapi.Query(None) - ) -> responses.ProcessList: + ) -> models.ProcessList: if not limit: limit = len(PROCESSES_DB) processes = [ - responses.ProcessSummary( + models.ProcessSummary( id=PROCESSES_DB[i]["id"], version=PROCESSES_DB[i]["version"], ) for i in range(0, limit) ] - process_list = responses.ProcessList(processes=processes) + process_list = models.ProcessList(processes=processes) return process_list def get_process( self, process_id: str = fastapi.Path(...) - ) -> responses.ProcessDescription: + ) -> models.ProcessDescription: for i, elem in enumerate(PROCESSES_DB): if elem["id"] == process_id: - process = responses.ProcessDescription( + process = models.ProcessDescription( id=PROCESSES_DB[i]["id"], version=PROCESSES_DB[i]["version"], inputs=PROCESSES_DB[i]["inputs"], @@ -68,8 +68,8 @@ def post_process_execution( self, process_id: str = fastapi.Path(...), execution_content: Dict[str, Any] = fastapi.Body(...), - ) -> responses.StatusInfo: - status_info = responses.StatusInfo(jobID=1, status="accepted", type="process") + ) -> models.StatusInfo: + status_info = models.StatusInfo(jobID=1, status="accepted", type="process") return status_info def get_jobs( @@ -77,13 +77,13 @@ def get_jobs( processID: Optional[List[str]] = fastapi.Query(None), status: Optional[List[str]] = fastapi.Query(None), limit: Optional[int] = fastapi.Query(10, ge=1, le=10000), - ) -> responses.JobList: - jobs = [responses.StatusInfo(jobID=1, status="accepted", type="process")] - job_list = responses.JobList(jobs=jobs) + ) -> models.JobList: + jobs = [models.StatusInfo(jobID=1, status="accepted", type="process")] + job_list = models.JobList(jobs=jobs) return job_list - def get_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: - status_info = responses.StatusInfo(jobID=1, status="running", type="process") + def get_job(self, job_id: str = fastapi.Path(...)) -> models.StatusInfo: + status_info = models.StatusInfo(jobID=1, status="running", type="process") return status_info def get_job_results( # type: ignore @@ -95,16 +95,16 @@ def get_job_results( # type: ignore } return results - def delete_job(self, job_id: str = fastapi.Path(...)) -> responses.StatusInfo: - status_info = responses.StatusInfo(jobID=1, status="dismissed", type="process") + def delete_job(self, job_id: str = fastapi.Path(...)) -> models.StatusInfo: + status_info = models.StatusInfo(jobID=1, status="dismissed", type="process") return status_info -class StatusInfo(responses.StatusInfo): +class StatusInfo(models.StatusInfo): metadata: str -class JobList(responses.JobList): +class JobList(models.JobList): jobs: List[StatusInfo] # type: ignore @@ -115,25 +115,25 @@ class TestClientExtended(clients.BaseClient): def get_processes( self, limit: Optional[int] = fastapi.Query(None) - ) -> responses.ProcessList: + ) -> models.ProcessList: if not limit: limit = len(PROCESSES_DB) processes = [ - responses.ProcessSummary( + models.ProcessSummary( id=PROCESSES_DB[i]["id"], version=PROCESSES_DB[i]["version"], ) for i in range(0, limit) ] - process_list = responses.ProcessList(processes=processes) + process_list = models.ProcessList(processes=processes) return process_list def get_process( self, process_id: str = fastapi.Path(...) - ) -> responses.ProcessDescription: + ) -> models.ProcessDescription: for i, elem in enumerate(PROCESSES_DB): if elem["id"] == process_id: - process = responses.ProcessDescription( + process = models.ProcessDescription( id=PROCESSES_DB[i]["id"], version=PROCESSES_DB[i]["version"], inputs=PROCESSES_DB[i]["inputs"], @@ -146,8 +146,8 @@ def post_process_execution( self, process_id: str = fastapi.Path(...), execution_content: Dict[str, Any] = fastapi.Body(...), - ) -> responses.StatusInfo: - status_info = responses.StatusInfo(jobID=1, status="accepted", type="process") + ) -> models.StatusInfo: + status_info = models.StatusInfo(jobID=1, status="accepted", type="process") return status_info def get_jobs( diff --git a/tests/test_10_main.py b/tests/test_10_main.py index 7afadb3..73595f4 100644 --- a/tests/test_10_main.py +++ b/tests/test_10_main.py @@ -37,7 +37,7 @@ def test_set_resp_model_get_landing_page_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetLandingPage" ) - exp_resp_model = ogc_api_processes_fastapi.responses.LandingPage + exp_resp_model = ogc_api_processes_fastapi.models.LandingPage resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) @@ -50,7 +50,7 @@ def test_set_resp_model_get_conformance_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetConformance" ) - exp_resp_model = ogc_api_processes_fastapi.responses.ConfClass + exp_resp_model = ogc_api_processes_fastapi.models.ConfClass resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) @@ -63,7 +63,7 @@ def test_set_resp_model_get_process_list_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetProcesses" ) - exp_resp_model = ogc_api_processes_fastapi.responses.ProcessList + exp_resp_model = ogc_api_processes_fastapi.models.ProcessList resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert resp_model_schema["properties"] == exp_resp_model_schema["properties"] @@ -80,7 +80,7 @@ def test_set_resp_model_get_process_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetProcess" ) - exp_resp_model = ogc_api_processes_fastapi.responses.ProcessDescription + exp_resp_model = ogc_api_processes_fastapi.models.ProcessDescription resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) @@ -93,7 +93,7 @@ def test_set_resp_model_post_process_execution_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "PostProcessExecution" ) - exp_resp_model = ogc_api_processes_fastapi.responses.StatusInfo + exp_resp_model = ogc_api_processes_fastapi.models.StatusInfo resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) @@ -106,7 +106,7 @@ def test_set_resp_model_get_jobs_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetJobs" ) - exp_resp_model = ogc_api_processes_fastapi.responses.JobList + exp_resp_model = ogc_api_processes_fastapi.models.JobList resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert resp_model_schema["properties"] == exp_resp_model_schema["properties"] @@ -123,7 +123,7 @@ def test_set_resp_model_get_job_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetJob" ) - exp_resp_model = ogc_api_processes_fastapi.responses.StatusInfo + exp_resp_model = ogc_api_processes_fastapi.models.StatusInfo resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) @@ -136,7 +136,7 @@ def test_set_resp_model_get_jobs_results_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "GetJobResults" ) - exp_resp_model = ogc_api_processes_fastapi.responses.Results + exp_resp_model = ogc_api_processes_fastapi.models.Results resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) @@ -149,7 +149,7 @@ def test_set_resp_model_delete_job_default( resp_model = ogc_api_processes_fastapi.main.set_response_model( test_client_default, "DeleteJob" ) - exp_resp_model = ogc_api_processes_fastapi.responses.StatusInfo + exp_resp_model = ogc_api_processes_fastapi.models.StatusInfo resp_model_schema = resp_model.schema() exp_resp_model_schema = exp_resp_model.schema() assert equal_dicts(resp_model_schema, exp_resp_model_schema, ["title"]) diff --git a/tests/test_20_endpoints.py b/tests/test_20_endpoints.py index 6ff8e04..40ae73b 100644 --- a/tests/test_20_endpoints.py +++ b/tests/test_20_endpoints.py @@ -19,7 +19,7 @@ import pytest import ogc_api_processes_fastapi -from ogc_api_processes_fastapi import endpoints, responses +from ogc_api_processes_fastapi import endpoints, models BASE_URL = "http://testserver/processes/" @@ -27,29 +27,27 @@ def test_create_self_link() -> None: request_url = "http://localhost/myapi" self_link = endpoints.create_self_link(request_url) - exp_link = responses.Link(href=request_url, rel="self") + exp_link = models.Link(href=request_url, rel="self") assert self_link == exp_link self_link = endpoints.create_self_link(request_url, title="Title", type="Type") - exp_link = responses.Link(href=request_url, rel="self", title="Title", type="Type") + exp_link = models.Link(href=request_url, rel="self", title="Title", type="Type") assert self_link == exp_link def test_create_page_link() -> None: request_url = "http://localhost/myapi" page = "next" - pagination_qs = responses.PaginationQueryParameters( + pagination_qs = models.PaginationQueryParameters( next={"cursor": "mycursor", "back": "True"} ) link_page = endpoints.create_page_link(request_url, page, pagination_qs) - exp_link = responses.Link( - href=f"{request_url}?cursor=mycursor&back=True", rel="next" - ) + exp_link = models.Link(href=f"{request_url}?cursor=mycursor&back=True", rel="next") assert link_page == exp_link request_url = "http://localhost/myapi" page = "previous" - pagination_qs = responses.PaginationQueryParameters( + pagination_qs = models.PaginationQueryParameters( next={"cursor": "mycursor", "back": "True"} ) with pytest.raises(ValueError): @@ -58,12 +56,12 @@ def test_create_page_link() -> None: def test_create_pagination_links() -> None: request_url = "http://localhost/myapi" - pagination_qs = responses.PaginationQueryParameters( + pagination_qs = models.PaginationQueryParameters( prev={"cursor": "mycursor", "back": "False"} ) pagination_links = endpoints.create_pagination_links(request_url, pagination_qs) exp_links = [ - responses.Link(href=f"{request_url}?cursor=mycursor&back=False", rel="prev") + models.Link(href=f"{request_url}?cursor=mycursor&back=False", rel="prev") ] assert pagination_links == exp_links