Skip to content

Commit

Permalink
test: init automatic tests for collection_from_items.py script TDE-1227
Browse files Browse the repository at this point in the history
… (#1012)

### Motivation

Having automatic tests for `collection_from_items.py` script.

### Modifications

- Refactor `collection_from_items.py` `main()` function so it can be
easily called passing a list of arguments.
- Add a simple test that verifies the `collection.json` file is created
on a usual use case.

### Verification

run `pytest`
  • Loading branch information
paulfouquet authored Jul 29, 2024
1 parent 61be096 commit e63cdbc
Show file tree
Hide file tree
Showing 2 changed files with 69 additions and 3 deletions.
12 changes: 9 additions & 3 deletions scripts/collection_from_items.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import argparse
import json
import os
from argparse import Namespace
from typing import List

import shapely.geometry
import shapely.ops
Expand All @@ -17,8 +19,7 @@
from scripts.stac.imagery.provider import Provider, ProviderRole


# pylint: disable=too-many-locals
def main() -> None:
def parse_args(args: List[str] | None) -> Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--uri", dest="uri", help="s3 path to items and collection.json write location", required=True)
parser.add_argument("--collection-id", dest="collection_id", help="Collection ID", required=True)
Expand Down Expand Up @@ -85,7 +86,12 @@ def main() -> None:
"--concurrency", dest="concurrency", help="The number of files to limit concurrent reads", required=True, type=int
)

arguments = parser.parse_args()
return parser.parse_args(args)


# pylint: disable=too-many-locals
def main(args: List[str] | None = None) -> None:
arguments = parse_args(args)
uri = arguments.uri

providers: list[Provider] = []
Expand Down
60 changes: 60 additions & 0 deletions scripts/tests/collection_from_items_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
from boto3 import client, resource
from moto import mock_aws
from moto.s3.responses import DEFAULT_REGION_NAME

from scripts.collection_from_items import main
from scripts.datetimes import utc_now
from scripts.files.fs_s3 import write
from scripts.json_codec import dict_to_json_bytes
from scripts.stac.imagery.item import ImageryItem


@mock_aws
def test_should_create_collection_file() -> None:
# Mock AWS S3
s3 = resource("s3", region_name=DEFAULT_REGION_NAME)
boto3_client = client("s3", region_name=DEFAULT_REGION_NAME)
s3.create_bucket(Bucket="stacfiles")
# Create mocked STAC Item
item = ImageryItem("123", "./scripts/tests/data/empty.tiff", utc_now)
geometry = {
"type": "Polygon",
"coordinates": [[1799667.5, 5815977.0], [1800422.5, 5815977.0], [1800422.5, 5814986.0], [1799667.5, 5814986.0]],
}
bbox = (1799667.5, 5815977.0, 1800422.5, 5814986.0)
start_datetime = "2021-01-27T00:00:00Z"
end_datetime = "2021-01-27T00:00:00Z"
item.update_spatial(geometry, bbox)
item.update_datetime(start_datetime, end_datetime)
item.add_collection("abc")
write("s3://stacfiles/item.json", dict_to_json_bytes(item.stac))
# CLI arguments
args = [
"--uri",
"s3://stacfiles/",
"--collection-id",
"abc",
"--category",
"urban-aerial-photos",
"--region",
"hawkes-bay",
"--gsd",
"1m",
"--start-date",
"2023-09-20",
"--end-date",
"2023-09-20",
"--lifecycle",
"ongoing",
"--producer",
"Placeholder",
"--licensor",
"Placeholder",
"--concurrency",
"25",
]
# Call script's main function
main(args)
# Verify collection.json has been created
resp = boto3_client.get_object(Bucket="stacfiles", Key="collection.json")
assert '"type": "Collection"' in resp["Body"].read().decode("utf-8")

0 comments on commit e63cdbc

Please sign in to comment.