test_api.py

"""Test WGC API base urls."""

import os

from pytest import fixture, raises

from requests import HTTPError

from webgeocalc import API, Api, ESA_API, JPL_API
from webgeocalc.errors import (APIError, APIResponseError, KernelSetNotFound,
                               ResultAttributeError, TooManyKernelSets)
from webgeocalc.vars import ESA_URL, JPL_URL


@fixture
def solar_system_kernel_set():
    """Solar kernels API output."""
    return {
        'caption': 'Solar System Kernels',
        'sclkId': '0',
        'description': 'Generic kernels for planets, satellites, and some asteroids, '
                       'covering at least from 1950-01-01 to 2050-01-01, and including '
                       'the latest generic text PCK and the mars_iau2000_v1.tpc PCK '
                       'loaded in this order.',
        'kernelSetId': '1',
        'missionId': 'gen',
    }


@fixture
def cassini_kernel_set():
    """Cassini kernel API output."""
    return {
        'caption': 'Cassini Huygens',
        'sclkId': '-82',
        'description': 'Archived CASSINI kernels covering from 1997-10-15 to 2017-09-15.',
        'kernelSetId': '5',
        'missionId': 'cassini',
    }


@fixture
def cassini_body():
    """Cassini body API output."""
    return {
        'id': -82,
        'name': 'CASSINI',
    }


@fixture
def cassini_frame():
    """Cassini frame API output."""
    return {
        'id': -82905,
        'name': 'CASSINI_KSO',
        'centerBodyID': 699,
        'frameClass': 5,
    }


@fixture
def cassini_instrument():
    """CIRS instrument API output."""
    return {
        'id': -82898,
        'name': 'CASSINI_CIRS_RAD',
    }


@fixture
def api_queued():
    """Queued API response."""
    return {
        "status": "OK",
        "message": "The request was successful.",
        "calculationId": "0788aba2-d4e5-4028-9ef1-4867ad5385e0",
        "result": {
            "phase": "QUEUED",
            "position": 6
        }
    }


@fixture
def api_error():
    """Error API response."""
    return {
        "status": "ERROR",
        "message": "The request has failed.",
        "calculationId": "0788aba2-d4e5-4028-9ef1-4867ad5385e0",
        "error": {
            "shortDescription": "The provided Calculation ID does not "
                                "correspond to any requested calculation."
        }
    }


@fixture
def api_empty_data_response():
    """Empty data API response."""
    return {
        "status": "OK",
        "message": "The operation was successful.",
        "calculationId": "0788aba2-d4e5-4028-9ef1-4867ad5385e0",
    }


def test_api_url():
    """Test default API url."""
    assert str(API) == API.url
    assert str(JPL_API) == JPL_URL
    assert str(ESA_API) == ESA_URL


def test_api_env_url(monkeypatch):
    """Test default API url from environment variable `WGC_URL`."""
    monkeypatch.setenv('WGC_URL', 'https://wgc.obspm.fr/webgeocalc/api')

    assert 'WGC_URL' in os.environ
    assert os.environ['WGC_URL'] == 'https://wgc.obspm.fr/webgeocalc/api'

    api = Api()
    assert str(api) == api.url == 'https://wgc.obspm.fr/webgeocalc/api'

    # Remove `WGC_URL`, fallback to JPL API
    monkeypatch.delenv('WGC_URL')

    assert 'WGC_URL' not in os.environ

    api = Api()
    assert str(api) == api.url == JPL_URL


def test_api_metadata():
    """Test API metadata."""
    assert API['title'] == 'WebGeocalc by NAIF'
    assert API['description'] == \
        'WGC2 -- a WebGeocalc Server with enabled API at NAIF, JPL'
    assert API['documentation'] == \
        'https://wgc2.jpl.nasa.gov:8443/webgeocalc/documents/api-info.html'
    assert API['contact'] == 'Boris Semenov <Boris.Semenov@jpl.nasa.gov>'
    assert API['version'] == '2.7.6'
    assert API['build_id'] == '5363 N67 29-JAN-2025'

    with raises(KeyError):
        _ = API['foo']


def test_response_err():
    """Test GET and POST on invalid URLs."""
    with raises(HTTPError):
        API.get('/foo')  # 404 error

    with raises(HTTPError):
        API.post('/', payload={})  # 404 error


def test_kernel_sets(solar_system_kernel_set):
    """Test GET kernel sets from API."""
    kernel_set = API.kernel_sets()[0]

    assert int(kernel_set) == int(solar_system_kernel_set['kernelSetId'])
    assert str(kernel_set) == solar_system_kernel_set['caption']

    for key in kernel_set.keys():
        assert key in solar_system_kernel_set.keys()

    for value in kernel_set.values():
        assert value in solar_system_kernel_set.values()

    for key, value in kernel_set.items():
        assert key in solar_system_kernel_set.keys()
        assert value in solar_system_kernel_set.values()

    with raises(ResultAttributeError):
        _ = kernel_set.wrong_attr


def test_kernel_set_by_id():
    """Test kernel set id."""
    assert int(API.kernel_set(1)) == 1


def test_kernel_set_by_full_name():
    """Test kernel set name."""
    assert str(API.kernel_set('Solar System Kernels')) == 'Solar System Kernels'


def test_kernel_set_by_name_not_case_sensitive():
    """Test case sensitivity of kernel set search."""
    assert str(API.kernel_set('solar system kernels')) == 'Solar System Kernels'


def test_kernel_set_by_name_partial():
    """Test partial kernel set search."""
    assert str(API.kernel_set('Solar')) == 'Solar System Kernels'


def test_kernel_set_too_many_found():
    """Test error if too many kernel sets are found."""
    with raises(TooManyKernelSets):
        assert str(API.kernel_set('Cassini'))


def test_kernel_set_not_found():
    """Test error if kernel set not found."""
    with raises(KernelSetNotFound):
        assert str(API.kernel_set('Missing kernel'))


def test_kernel_set_id_for_str(cassini_kernel_set):
    """Test kernel set search for id and error cases."""
    kernel_set_id = int(cassini_kernel_set['kernelSetId'])
    kernel_set_caption = cassini_kernel_set['caption']

    assert API.kernel_set_id(kernel_set_caption) == kernel_set_id

    kernel_set = API.kernel_set(kernel_set_caption)
    assert API.kernel_set_id(kernel_set) == kernel_set_id

    with raises(TypeError):
        API.kernel_set_id(1.23)


def test_bodies(cassini_kernel_set, cassini_body):
    """Test GET bodies from API."""
    kernel_set_id = int(cassini_kernel_set['kernelSetId'])
    body = API.bodies(kernel_set_id)[0]

    assert int(body) == cassini_body['id']
    assert str(body) == cassini_body['name']


def test_frames(cassini_kernel_set, cassini_frame):
    """Test GET frames from API."""
    kernel_set_id = int(cassini_kernel_set['kernelSetId'])
    frame = API.frames(kernel_set_id)[58]

    assert int(frame) == cassini_frame['id']
    assert str(frame) == cassini_frame['name']

    for value in frame.values():
        assert value in cassini_frame.values()


def test_instruments(cassini_kernel_set, cassini_instrument):
    """Test GET instruments from API."""
    kernel_set_id = int(cassini_kernel_set['kernelSetId'])
    instrument = API.instruments(kernel_set_id)[0]

    assert int(instrument) == cassini_instrument['id']
    assert str(instrument) == cassini_instrument['name']


def test_api_read_queued(api_queued):
    """Test queued position output from API."""
    _, phase = API.read(api_queued)
    assert phase == 'QUEUED | POSITION: 6'


def test_api_read_error(api_error):
    """Test error output from API."""
    with raises(APIError):
        API.read(api_error)


def test_api_read_invalid(api_empty_data_response):
    """Test error if response no valid data."""
    with raises(APIResponseError):
        API.read(api_empty_data_response)