|
13 | 13 | OpenEOPlatform, |
14 | 14 | ) |
15 | 15 | from app.schemas.enum import OutputFormatEnum, ProcessingStatusEnum |
| 16 | +from app.schemas.parameters import ParamTypeEnum, Parameter |
16 | 17 | from app.schemas.unit_job import ServiceDetails |
17 | 18 | from stac_pydantic import Collection |
18 | 19 |
|
@@ -566,3 +567,88 @@ async def test_execute_sync_job_success( |
566 | 567 | assert response.status_code == mock_response.status_code |
567 | 568 | assert json.loads(response.body) == json.loads(mock_response.content) |
568 | 569 | mock_connect.assert_called_once_with("fake_token", service_details.endpoint) |
| 570 | + |
| 571 | + |
| 572 | +@pytest.mark.asyncio |
| 573 | +@patch("app.platforms.implementations.openeo.requests.get") |
| 574 | +async def test_get_parameters_success(mock_udp_request, platform): |
| 575 | + |
| 576 | + udp_params = [ |
| 577 | + { |
| 578 | + "name": "flag_test", |
| 579 | + "description": "Test for a boolean flag parameter", |
| 580 | + "schema": {"type": "boolean"}, |
| 581 | + }, |
| 582 | + { |
| 583 | + "name": "bbox_test", |
| 584 | + "description": "Test for a bbox parameter", |
| 585 | + "schema": {"type": "object", "subtype": "bounding-box"}, |
| 586 | + }, |
| 587 | + { |
| 588 | + "name": "date_test", |
| 589 | + "description": "Test for a date parameter", |
| 590 | + "schema": {"type": "array", "subtype": "temporal-interval"}, |
| 591 | + "optional": True, |
| 592 | + "default": ["2020-01-01", "2020-12-31"], |
| 593 | + }, |
| 594 | + ] |
| 595 | + mock_udp_request.return_value.json.return_value = { |
| 596 | + "id": "process123", |
| 597 | + "parameters": udp_params, |
| 598 | + } |
| 599 | + mock_udp_request.return_value.raise_for_status.return_value = None |
| 600 | + result = await platform.get_service_parameters( |
| 601 | + user_token="fake_token", |
| 602 | + details=ServiceDetails( |
| 603 | + endpoint="https://openeo.dataspace.copernicus.eu", |
| 604 | + application="https://foo.bar/process.json", |
| 605 | + ), |
| 606 | + ) |
| 607 | + parameters = [ |
| 608 | + Parameter( |
| 609 | + name=udp_params[0]["name"], |
| 610 | + description=udp_params[0]["description"], |
| 611 | + type=ParamTypeEnum.BOOLEAN, |
| 612 | + optional=False, |
| 613 | + ), |
| 614 | + Parameter( |
| 615 | + name=udp_params[1]["name"], |
| 616 | + description=udp_params[1]["description"], |
| 617 | + type=ParamTypeEnum.BOUNDING_BOX, |
| 618 | + optional=False, |
| 619 | + ), |
| 620 | + Parameter( |
| 621 | + name=udp_params[2]["name"], |
| 622 | + description=udp_params[2]["description"], |
| 623 | + type=ParamTypeEnum.DATE_INTERVAL, |
| 624 | + optional=True, |
| 625 | + default=udp_params[2]["default"], |
| 626 | + ), |
| 627 | + ] |
| 628 | + assert result == parameters |
| 629 | + |
| 630 | + |
| 631 | +@pytest.mark.asyncio |
| 632 | +@patch("app.platforms.implementations.openeo.requests.get") |
| 633 | +async def test_get_parameters_unsupported_type(mock_udp_request, platform): |
| 634 | + |
| 635 | + mock_udp_request.return_value.json.return_value = { |
| 636 | + "id": "process123", |
| 637 | + "parameters": [ |
| 638 | + { |
| 639 | + "name": "foobar_test", |
| 640 | + "description": "Test for a foobar parameter", |
| 641 | + "schema": {"type": "foobar"}, |
| 642 | + } |
| 643 | + ], |
| 644 | + } |
| 645 | + mock_udp_request.return_value.raise_for_status.return_value = None |
| 646 | + |
| 647 | + with pytest.raises(ValueError, match="Unsupported parameter schemas"): |
| 648 | + await platform.get_service_parameters( |
| 649 | + user_token="fake_token", |
| 650 | + details=ServiceDetails( |
| 651 | + endpoint="https://openeo.dataspace.copernicus.eu", |
| 652 | + application="https://foo.bar/process.json", |
| 653 | + ), |
| 654 | + ) |
0 commit comments