sds-data-manager/tests/lambda_endpoints/test_ialirt_catalog_api.py at d99c706797d02ef24e035986321a863573b4df75 · laspsandoval/sds-data-manager · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
"""Tests for the I-ALiRT Catalog API."""

import json

from sds_data_manager.lambda_code.IAlirtCode import ialirt_catalog_api


def test_bad_dates_ialirt_catalog_api(s3_client):
    """Test a failing call to the catalog endpoint due to bad dates."""
    event = {
        "queryStringParameters": {
            "start_date": "2025-02-06",
            "end_date": "2025-02-06",
            "station": "station1",
        }
    }
    response = ialirt_catalog_api.lambda_handler(event=event, context=None)
    assert response["statusCode"] == 400
    assert response["body"] == (
        "The start date of the requested time frame is not "
        "before the end date. Please supply a valid time "
        "range."
    )


def test_too_large_time_range_ialirt_catalog_api(s3_client):
    """Test a failing call to the endpoint caused by dates that are too far apart."""
    event = {
        "queryStringParameters": {
            "start_date": "2025-02-06",
            "end_date": "2025-03-20",
            "station": "station1",
        }
    }
    response = ialirt_catalog_api.lambda_handler(event=event, context=None)
    assert response["statusCode"] == 400
    assert response["body"] == (
        "The end date of the requested time frame must not be "
        "more than 30 days after the start date. Please supply "
        "a valid time range."
    )


def test_no_files_ialirt_catalog_api(s3_client):
    """Test unsuccessful call to the endpoint caused by an empty response from s3."""
    event = {
        "queryStringParameters": {
            "start_date": "2025-02-06",
            "end_date": "2025-02-07",
            "station": "station1",
        }
    }
    response = ialirt_catalog_api.lambda_handler(event=event, context=None)
    assert response["statusCode"] == 404


def test_success_ialirt_catalog_api(s3_client):
    """Test a successful call to the catalog endpoint."""
    test_file1 = "pointing_schedules/station1/2025-02-06_station1_01.txt"
    test_file2 = "pointing_schedules/station1/2025-02-06_station1_02.txt"
    s3_client.put_object(
        Bucket="test-data-bucket",
        Key=test_file1,
        Body=b"Hello world 1",
    )
    s3_client.put_object(
        Bucket="test-data-bucket",
        Key=test_file2,
        Body=b"Hello world 2",
    )
    event = {
        "queryStringParameters": {
            "start_date": "2025-02-06",
            "end_date": "2025-02-07",
            "station": "station1",
        }
    }
    response = ialirt_catalog_api.lambda_handler(event=event, context=None)
    assert response["statusCode"] == 200
    assert json.loads(response["body"])["file_names"] == [
        "2025-02-06_station1_01.txt",
        "2025-02-06_station1_02.txt",
    ]