amancevice/terraform-aws-serverless-pypi

import io
import json
import os
import re
from unittest import mock

import pytest

os.environ["S3_BUCKET"] = "serverless-pypi"

with mock.patch("boto3.client"):
    import index
    from index import ANCHOR, INDEX

SIMPLE_INDEX = INDEX.safe_substitute(
    title="Simple index",
    anchors=str.join(
        "",
        [
            ANCHOR.safe_substitute(href="fizz", name="fizz"),
            ANCHOR.safe_substitute(href="buzz", name="buzz"),
        ],
    ),
)
PACKAGE_INDEX = INDEX.safe_substitute(
    title="Links for fizz",
    anchors=str.join(
        "",
        [
            ANCHOR.safe_substitute(href="presigned-url", name="fizz-0.1.2.tar.gz"),
            ANCHOR.safe_substitute(href="presigned-url", name="fizz-1.2.3.tar.gz"),
        ],
    ),
)
S3_REINDEX_RESPONSE = [
    {"CommonPrefixes": [{"Prefix": "fizz/"}, {"Prefix": "buzz/"}]},
]
S3_INDEX_RESPONSE = [
    {
        "Contents": [
            {"Key": "fizz/fizz-0.1.2.tar.gz"},
            {"Key": "fizz/fizz-1.2.3.tar.gz"},
        ],
    },
]


def test_proxy_reponse():
    body = "FIZZ"
    ret = index.proxy_reponse("FIZZ")
    exp = {
        "body": body,
        "statusCode": 200,
        "headers": {
            "content-length": len(body),
            "content-type": "text/html; charset=utf-8",
        },
    }
    assert ret == exp


def test_get_index():
    index.S3.get_object.return_value = {
        "Body": io.BytesIO(SIMPLE_INDEX.encode()),
    }
    ret = index.get_index()
    exp = {
        "body": SIMPLE_INDEX,
        "statusCode": 200,
        "headers": {
            "content-length": len(SIMPLE_INDEX),
            "content-type": "text/html; charset=utf-8",
        },
    }
    assert ret == exp


def test_get_package_index():
    index.S3.generate_presigned_url.return_value = "presigned-url"
    index.S3_PAGINATOR.paginate.return_value = iter(S3_INDEX_RESPONSE)
    ret = index.get_package_index("fizz")
    exp = {
        "body": PACKAGE_INDEX,
        "statusCode": 200,
        "headers": {
            "content-length": len(PACKAGE_INDEX),
            "content-type": "text/html; charset=utf-8",
        },
    }
    assert ret == exp


def test_get_package_index_fallback():
    index.FALLBACK_INDEX_URL = "https://pypi.org/simple/"
    index.S3_PAGINATOR.paginate.return_value = iter([])
    ret = index.get_package_index("buzz")
    exp = {
        "statusCode": 301,
        "headers": {
            "Location": "https://pypi.org/simple/buzz/",
        },
    }
    assert ret == exp


def test_get_package_index_not_found():
    index.FALLBACK_INDEX_URL = ""
    index.S3_PAGINATOR.paginate.return_value = iter([])
    body = json.dumps({"message": "Not Found"})
    ret = index.get_package_index("buzz")
    exp = {
        "body": body,
        "statusCode": 404,
        "headers": {
            "content-length": len(body),
            "content-type": "application/json; charset=utf-8",
        },
    }
    assert ret == exp


def test_redirect():
    ret = index.redirect("simple")
    exp = {"headers": {"Location": "simple"}, "statusCode": 301}
    assert ret == exp


def test_reject():
    body = json.dumps({"message": "Unauthorized"})
    ret = index.reject(401, message="Unauthorized")
    exp = {
        "body": body,
        "statusCode": 401,
        "headers": {
            "content-length": len(body),
            "content-type": "application/json; charset=utf-8",
        },
    }
    assert ret == exp


@pytest.mark.parametrize(
    ("event", "exp"),
    [
        (
            {"version": "2.0", "routeKey": "GET /"},
            {
                "statusCode": 200,
                "body": "",
                "headers": {
                    "content-length": 0,
                    "content-type": "text/html; charset=utf-8",
                },
            },
        ),
    ],
)
def test_handler_get_root(event, exp):
    ret = index.proxy_request(event)
    assert ret == exp


@pytest.mark.parametrize(
    "event",
    [
        {"version": "2.0", "routeKey": "GET /"},
        {"version": "2.0", "routeKey": "HEAD /"},
        {"httpMethod": "GET"},
        {"httpMethod": "HEAD"},
    ],
)
def test_proxy_request_get(event):
    with mock.patch("index.get_index") as mock_idx:
        mock_idx.return_value = index.proxy_reponse(SIMPLE_INDEX)
        index.proxy_request(event)
        mock_idx.assert_called_once_with()


@pytest.mark.parametrize(
    "event",
    [
        {
            "version": "2.0",
            "routeKey": "GET /fizz",
            "pathParameters": {"package": "fizz"},
        },
        {"httpMethod": "GET", "pathParameters": {"package": "fizz"}},
    ],
)
def test_proxy_request_get_package(event):
    with mock.patch("index.get_package_index") as mock_pkg:
        mock_pkg.return_value = index.proxy_reponse(PACKAGE_INDEX)
        index.proxy_request(event)
        mock_pkg.assert_called_once_with("fizz")


@pytest.mark.parametrize(
    ("event", "status_code", "msg"),
    [
        (
            {
                "version": "2.0",
                "routeKey": "OPTIONS /fizz",
                "pathParameters": {"package": "fizz"},
            },
            403,
            "Forbidden",
        ),
        (
            {
                "version": "2.0",
                "routeKey": "POST /fizz",
                "pathParameters": {"package": "fizz"},
            },
            405,
            "Not Allowed",
        ),
    ],
)
def test_proxy_request_reject(event, status_code, msg):
    ret = index.proxy_request(event)
    exp = index.reject(status_code, message=msg)
    assert ret == exp


def test_reindex_bucket():
    index.S3_PAGINATOR.paginate.return_value = iter(S3_REINDEX_RESPONSE)
    index.reindex_bucket({})
    index.S3.put_object.assert_called_once_with(
        Bucket=index.S3_BUCKET,
        Key="index.html",
        Body=SIMPLE_INDEX.encode(),
    )