1#
2# Copyright (C) 2023 The Android Open Source Project
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#      http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16"""Tests for fetchartifact."""
17from typing import cast
18
19import pytest
20from aiohttp import ClientResponseError, ClientSession
21from aiohttp.test_utils import TestClient
22from aiohttp.web import Application, Request, Response
23
24from fetchartifact import fetch_artifact, fetch_artifact_chunked
25
26TEST_BUILD_ID = "1234"
27TEST_TARGET = "linux"
28TEST_ARTIFACT_NAME = "output.zip"
29TEST_DOWNLOAD_URL = (
30    f"/android/internal/build/v3/builds/{TEST_BUILD_ID}/{TEST_TARGET}/"
31    f"attempts/latest/artifacts/{TEST_ARTIFACT_NAME}/url"
32)
33TEST_RESPONSE = b"Hello, world!"
34
35
36@pytest.fixture(name="android_ci_client")
37async def fixture_android_ci_client(aiohttp_client: type[TestClient]) -> TestClient:
38    """Fixture for mocking the Android CI APIs."""
39
40    async def download(_request: Request) -> Response:
41        return Response(text=TEST_RESPONSE.decode("utf-8"))
42
43    app = Application()
44    app.router.add_get(TEST_DOWNLOAD_URL, download)
45    return await aiohttp_client(app)  # type: ignore
46
47
48async def test_fetch_artifact(android_ci_client: TestClient) -> None:
49    """Tests that the download URL is queried."""
50    assert TEST_RESPONSE == await fetch_artifact(
51        TEST_TARGET,
52        TEST_BUILD_ID,
53        TEST_ARTIFACT_NAME,
54        cast(ClientSession, android_ci_client),
55        query_url_base="",
56    )
57
58
59async def test_fetch_artifact_chunked(android_ci_client: TestClient) -> None:
60    """Tests that the full file contents are downloaded."""
61    assert [c.encode("utf-8") for c in TEST_RESPONSE.decode("utf-8")] == [
62        chunk
63        async for chunk in fetch_artifact_chunked(
64            TEST_TARGET,
65            TEST_BUILD_ID,
66            TEST_ARTIFACT_NAME,
67            cast(ClientSession, android_ci_client),
68            chunk_size=1,
69            query_url_base="",
70        )
71    ]
72
73
74async def test_failure_raises(android_ci_client: TestClient) -> None:
75    """Tests that fetch failure raises an exception."""
76    with pytest.raises(ClientResponseError):
77        await fetch_artifact(
78            TEST_TARGET,
79            TEST_BUILD_ID,
80            TEST_ARTIFACT_NAME,
81            cast(ClientSession, android_ci_client),
82            query_url_base="/bad",
83        )
84
85    with pytest.raises(ClientResponseError):
86        async for _chunk in fetch_artifact_chunked(
87            TEST_TARGET,
88            TEST_BUILD_ID,
89            TEST_ARTIFACT_NAME,
90            cast(ClientSession, android_ci_client),
91            query_url_base="/bad",
92        ):
93            pass
94
95
96@pytest.mark.requires_network
97async def test_real_artifact() -> None:
98    """Tests with a real artifact. Requires an internet connection."""
99    async with ClientSession() as session:
100        contents = await fetch_artifact("linux", "9945621", "logs/SUCCEEDED", session)
101        assert contents == b"1681499053\n"
102