"""Tests for the STS Connector.""" import datetime from unittest import mock from freezegun import freeze_time from pytest_mock import MockerFixture from video.connectors import sts from video.constants import job_io_fields, job_types from video.constants.thumbnails import UPLOAD_CUSTOM_THUMBNAIL from video.models.s3 import s3_token @freeze_time("2019-05-11 09:50:00") def test_get_s3_token(mocker: MockerFixture) -> None: """Test credentials returned.""" client_mock = mock.MagicMock() client_mock.assume_role.return_value = { "Credentials": { "SessionToken": "session", "AccessKeyId": "access", "SecretAccessKey": "secret", "Expiration": datetime.datetime.now(datetime.timezone.utc), } } mocker.patch.object(sts, "connect_to_sts", return_value=client_mock) result = s3_token.get_s3_token( 55, job_types.TRANSFER_FROM_BROWSER_TO_S3, "filename.mov" ) assert result == { job_io_fields.AWS_ACCESS_KEY: "access", job_io_fields.AWS_SECRET_ACCESS_KEY: "secret", job_io_fields.TOKEN: "session", job_io_fields.INPUT_VIDEO_S3_BUCKET: "test-orcd-video-assets", job_io_fields.INPUT_VIDEO_S3_KEY: "raw/55/filename.mov", job_io_fields.S3_TOKEN_EXPIRATION: "2019-05-11 09:50:00", } @freeze_time("2019-05-11 09:50:00") def test_get_s3_token_custom_thumbnail(mocker: MockerFixture) -> None: """Test credentials returned.""" client_mock = mock.MagicMock() client_mock.assume_role.return_value = { "Credentials": { "SessionToken": "session", "AccessKeyId": "access", "SecretAccessKey": "secret", "Expiration": datetime.datetime.now(datetime.timezone.utc), } } mocker.patch.object(sts, "connect_to_sts", return_value=client_mock) result = s3_token.get_s3_token(123, UPLOAD_CUSTOM_THUMBNAIL, "123.jpg") assert result == { job_io_fields.AWS_ACCESS_KEY: "access", job_io_fields.AWS_SECRET_ACCESS_KEY: "secret", job_io_fields.TOKEN: "session", job_io_fields.INPUT_VIDEO_S3_BUCKET: "test-orcd-video-assets", job_io_fields.INPUT_VIDEO_S3_KEY: "thumbnails/custom/123/123.jpg", job_io_fields.S3_TOKEN_EXPIRATION: "2019-05-11 09:50:00", }