FabledGrimoire/tests/test_services_backup_s3.py

"""The backup destination's S3 client (services/backup_s3.py): signatures match AWS's
published Signature Version 4 examples, and the calls read S3's answers."""
import asyncio
from datetime import datetime, timezone

import httpx
import pytest

from grimoire.services import backup_s3 as s3

# AWS's example credentials and moment ("Signature Calculations for the
# Authorization Header", Amazon S3 API reference).
KEY_ID = "AKIAIOSFODNN7EXAMPLE"
SECRET = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
AT = datetime(2013, 5, 24, tzinfo=timezone.utc)


def _signature(headers):
    return headers["Authorization"].rsplit("Signature=", 1)[1]


def test_aws_example_get_object():
    h = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/test.txt", {"Range": "bytes=0-9"},
                s3.EMPTY_SHA256, KEY_ID, SECRET, "us-east-1", AT)
    assert "SignedHeaders=host;range;x-amz-content-sha256;x-amz-date," in h["Authorization"]
    assert "Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request" in h["Authorization"]
    assert _signature(h) == "f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"


def test_aws_example_list_objects():
    h = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/?max-keys=2&prefix=J", {},
                s3.EMPTY_SHA256, KEY_ID, SECRET, "us-east-1", AT)
    assert _signature(h) == "34b48302e7b5fa45bde8084f4b7868a86f0a534bc59db6670ed5711ef69dc6f7"


def test_a_changed_secret_changes_the_signature():
    a = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/test.txt", {},
                s3.EMPTY_SHA256, KEY_ID, SECRET, "us-east-1", AT)
    b = s3.sign("GET", "https://examplebucket.s3.amazonaws.com/test.txt", {},
                s3.EMPTY_SHA256, KEY_ID, SECRET + "x", "us-east-1", AT)
    assert _signature(a) != _signature(b)


TARGET = s3.S3Target("http://archives:8081/s3", "grimoire", KEY_ID, SECRET)


def test_paths_are_path_style():
    assert TARGET.url("grimoire-20261012T000000Z.tar") == "http://archives:8081/s3/grimoire/grimoire-20261012T000000Z.tar"
    assert TARGET.url() == "http://archives:8081/s3/grimoire/"


def test_put_sends_the_file_signed(tmp_path):
    f = tmp_path / "b.tar"
    f.write_bytes(b"x" * 5000)
    seen = {}

    def handler(request: httpx.Request):
        seen["method"], seen["url"] = request.method, str(request.url)
        seen["body"] = request.read()
        seen["auth"] = request.headers["authorization"]
        seen["sha"] = request.headers["x-amz-content-sha256"]
        return httpx.Response(200)

    asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=httpx.MockTransport(handler)))
    assert seen["method"] == "PUT"
    assert seen["url"] == "http://archives:8081/s3/grimoire/b.tar"
    assert seen["body"] == b"x" * 5000
    assert seen["sha"] == s3.file_sha256(f)
    assert seen["auth"].startswith("AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/")


def test_a_refusal_says_why(tmp_path):
    f = tmp_path / "b.tar"
    f.write_bytes(b"x")
    body = b"<Error><Code>SignatureDoesNotMatch</Code><Message>no</Message></Error>"
    transport = httpx.MockTransport(lambda r: httpx.Response(403, content=body))
    with pytest.raises(s3.S3Error, match="HTTP 403 SignatureDoesNotMatch: the secret key is wrong"):
        asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=transport))


def test_a_web_server_at_the_address_is_named_as_not_s3():
    """No port in the address: the request lands on whatever serves port 80,
    and its plain 404 says nothing about S3."""
    page = b"<html><body><h1>404 Not Found</h1></body></html>"
    transport = httpx.MockTransport(lambda r: httpx.Response(404, content=page))
    with pytest.raises(s3.S3Error, match=r"HTTP 404: the address answered, but not as S3 — check the scheme and the port"):
        asyncio.run(s3.list_objects(TARGET, transport=transport))


def test_a_server_error_without_a_body_is_not_blamed_on_the_port():
    transport = httpx.MockTransport(lambda r: httpx.Response(503))
    with pytest.raises(s3.S3Error, match="HTTP 503: Service Unavailable") as err:
        asyncio.run(s3.list_objects(TARGET, transport=transport))
    assert "port" not in str(err.value)


def test_unreachable_says_so(tmp_path):
    f = tmp_path / "b.tar"
    f.write_bytes(b"x")

    def handler(request):
        raise httpx.ConnectError("refused", request=request)

    with pytest.raises(s3.S3Error, match="Couldn't reach http://archives:8081/s3"):
        asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=httpx.MockTransport(handler)))


def test_list_reads_contents():
    xml = b"""<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
  <Name>grimoire</Name><KeyCount>2</KeyCount>
  <Contents><Key>grimoire-20261011T000000Z.tar</Key><Size>1200</Size>
    <LastModified>2026-10-11T00:00:01.000Z</LastModified></Contents>
  <Contents><Key>grimoire-20261012T000000Z.tar</Key><Size>1300</Size>
    <LastModified>2026-10-12T00:00:01.000Z</LastModified></Contents>
</ListBucketResult>"""
    seen = {}

    def handler(request):
        seen["params"] = dict(request.url.params)
        return httpx.Response(200, content=xml)

    got = asyncio.run(s3.list_objects(TARGET, "grimoire-", transport=httpx.MockTransport(handler)))
    assert seen["params"] == {"list-type": "2", "max-keys": "1000", "prefix": "grimoire-"}
    assert got == [
        s3.StoredObject("grimoire-20261011T000000Z.tar", 1200, "2026-10-11T00:00:01.000Z"),
        s3.StoredObject("grimoire-20261012T000000Z.tar", 1300, "2026-10-12T00:00:01.000Z"),
    ]


def test_get_streams_the_object_to_disk(tmp_path):
    seen = {}

    def handler(request):
        seen["method"], seen["url"] = request.method, str(request.url)
        seen["sha"] = request.headers["x-amz-content-sha256"]
        return httpx.Response(200, content=b"y" * 3000)

    out = tmp_path / "b.tar"
    asyncio.run(s3.get_file(TARGET, "b.tar", out, transport=httpx.MockTransport(handler)))
    assert seen["method"] == "GET"
    assert seen["url"] == "http://archives:8081/s3/grimoire/b.tar"
    assert seen["sha"] == s3.EMPTY_SHA256
    assert out.read_bytes() == b"y" * 3000


def test_a_refused_get_says_so_and_writes_nothing(tmp_path):
    transport = httpx.MockTransport(lambda r: httpx.Response(403, content=b"<Error><Code>AccessDenied</Code></Error>"))
    out = tmp_path / "b.tar"
    with pytest.raises(s3.S3Error, match="b.tar"):
        asyncio.run(s3.get_file(TARGET, "b.tar", out, transport=transport))
    assert not out.exists()