Ai/location_data/test_openeo_service.py

from decimal import Decimal
from io import StringIO
import os
from unittest.mock import Mock, patch

from django.core.management import call_command
from django.test import SimpleTestCase

from config.proxy import resolve_requests_proxy_url
from location_data.openeo_service import (
    OpenEOConnectionSettings,
    _resolve_openeo_proxy_url_from_env,
    build_empty_metric_payload,
    connect_openeo,
    is_openeo_auth_configured,
    linear_to_db,
    merge_metric_results,
    parse_aggregate_spatial_response,
)


class OpenEOServiceParsingTests(SimpleTestCase):
    def test_parse_feature_collection_results(self):
        payload = {
            "type": "FeatureCollection",
            "features": [
                {
                    "type": "Feature",
                    "id": "cell-1",
                    "properties": {"mean": 0.61},
                },
                {
                    "type": "Feature",
                    "id": "cell-2",
                    "properties": {"mean": 0.47},
                },
            ],
        }

        result = parse_aggregate_spatial_response(payload, "ndvi")

        self.assertEqual(result["cell-1"]["ndvi"], 0.61)
        self.assertEqual(result["cell-2"]["ndvi"], 0.47)

    def test_parse_mapping_results(self):
        payload = {
            "cell-1": {"mean": 12.4},
            "cell-2": {"mean": 15.1},
        }

        result = parse_aggregate_spatial_response(payload, "lst_c")

        self.assertEqual(result["cell-1"]["lst_c"], 12.4)
        self.assertEqual(result["cell-2"]["lst_c"], 15.1)

    def test_linear_to_db(self):
        self.assertEqual(linear_to_db(10.0), 10.0)
        self.assertEqual(linear_to_db(Decimal("1.0")), 0.0)
        self.assertIsNone(linear_to_db(0))
        self.assertIsNone(linear_to_db(-1))

    def test_merge_metric_results(self):
        target = {"cell-1": build_empty_metric_payload()}

        merge_metric_results(
            target,
            {
                "cell-1": {"ndvi": 0.5},
                "cell-2": {"ndwi": 0.2},
            },
        )

        self.assertEqual(target["cell-1"]["ndvi"], 0.5)
        self.assertEqual(target["cell-2"]["ndwi"], 0.2)
        self.assertIn("soil_vv_db", target["cell-2"])


class OpenEOConnectionTests(SimpleTestCase):
    def test_default_openeo_proxy_url_uses_proxychains_endpoint_without_wrapping_process(self):
        with patch.dict(
            os.environ,
            {
                "ENABLE_PROXYCHAINS": "0",
                "PROXYCHAINS_PROXY_TYPE": "socks4",
                "PROXYCHAINS_PROXY_HOST": "host.docker.internal",
                "PROXYCHAINS_PROXY_PORT": "10808",
                "OPENEO_PROXY_URL": "socks5h://host.docker.internal:10808",
            },
            clear=False,
        ):
            self.assertEqual(
                _resolve_openeo_proxy_url_from_env(),
                "socks4a://host.docker.internal:10808",
            )

    def test_requests_proxy_is_disabled_when_proxychains_targets_same_endpoint(self):
        with patch.dict(
            os.environ,
            {
                "ENABLE_PROXYCHAINS": "1",
                "PROXYCHAINS_PROXY_TYPE": "socks4",
                "PROXYCHAINS_PROXY_HOST": "host.docker.internal",
                "PROXYCHAINS_PROXY_PORT": "10808",
            },
            clear=False,
        ):
            self.assertEqual(
                resolve_requests_proxy_url("socks5h://host.docker.internal:10808"),
                "",
            )

    def test_is_openeo_auth_configured_for_client_credentials(self):
        self.assertTrue(
            is_openeo_auth_configured(
                OpenEOConnectionSettings(
                    auth_method="client_credentials",
                    client_id="client-id",
                    client_secret="client-secret",
                )
            )
        )

    def test_is_openeo_auth_configured_for_password(self):
        self.assertTrue(
            is_openeo_auth_configured(
                OpenEOConnectionSettings(
                    auth_method="password",
                    username="user@example.com",
                    password="secret",
                )
            )
        )

    def test_verify_openeo_auth_command_skips_when_unconfigured(self):
        stdout = StringIO()

        call_command("verify_openeo_auth", "--skip-if-unconfigured", stdout=stdout)

        self.assertIn("openEO auth check skipped", stdout.getvalue())

    def test_connect_openeo_applies_proxy_to_session(self):
        connection = Mock()
        connection.authenticate_oidc_resource_owner_password_credentials.return_value = connection
        openeo_module = Mock()
        openeo_module.connect.return_value = connection

        settings = OpenEOConnectionSettings(
            backend_url="https://openeofed.dataspace.copernicus.eu",
            auth_method="password",
            timeout_seconds=123,
            username="user@example.com",
            password="secret",
            proxy_url="socks5h://127.0.0.1:10808",
        )
        with patch.dict("sys.modules", {"openeo": openeo_module}):
            connect_openeo(settings)

        self.assertEqual(openeo_module.connect.call_args.kwargs["default_timeout"], 123)
        session = openeo_module.connect.call_args.kwargs["session"]
        self.assertEqual(session.proxies["https"], "socks5h://127.0.0.1:10808")
        self.assertFalse(session.trust_env)