system-config/testinfra/test_mirror.py

# Copyright 2019 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.


import json


testinfra_hosts = ['mirror01.openafs.provider.opendev.org',
                   'mirror02.openafs.provider.opendev.org']

def test_apache(host):
    apache = host.service('apache2')
    assert apache.is_running

def _run_cmd(host, port, scheme='https', url='', curl_opt=''):
    hostname = host.backend.get_hostname()
    return f'curl {curl_opt} --resolve {hostname}:127.0.0.1 {scheme}://{hostname}:{port}{url}'

def test_base_mirror(host):
    # base mirror
    cmd = host.run(_run_cmd(host, 443))
    assert '<a href="debian/">' in cmd.stdout

    # mirrors still respond on http
    cmd = host.run(_run_cmd(host, 80, scheme='http'))
    assert '<a href="debian/">' in cmd.stdout

def test_proxy_mirror(host):
    # pypi proxy mirror
    cmd = host.run(_run_cmd(host, 4443, url='/pypi/simple/setuptools'))
    assert 'setuptools' in cmd.stdout

    cmd = host.run(_run_cmd(host, 8080, scheme='http', url='/pypi/simple/setuptools'))
    assert 'setuptools' in cmd.stdout

def test_dockerv2_mirror(host):
    # Docker v2 mirror

    # NOTE(ianw) 2022-07 : this gets back a 401 .json; maybe something
    # better we could do?
    cmd = host.run(_run_cmd(host, 4445, url='/v2/'))
    assert 'UNAUTHORIZED' in cmd.stdout

    cmd = host.run(_run_cmd(host, 8082, scheme='http', url='/v2/'))
    assert 'UNAUTHORIZED' in cmd.stdout

def test_quay_mirror(host):
    # QuayRegistryMirror
    cmd = host.run(_run_cmd(host, 4447, url='/'))
    assert 'Quay' in cmd.stdout

    cmd = host.run(_run_cmd(host, 8084, scheme='http', url='/'))
    assert 'Quay' in cmd.stdout

# TODO test RHRegistryMirror

def test_galaxy_mirror(host):
    cmd = host.run(_run_cmd(host, 4448, url='/'))
    assert 'Ansible Galaxy' in cmd.stdout

    cmd = host.run(_run_cmd(host, 8085, scheme='http', url='/'))
    assert 'Ansible Galaxy' in cmd.stdout

    hostname = host.backend.get_hostname()
    # Ensure API properly answers
    cmd = host.run(_run_cmd(host, 4448, url='/api/'))
    assert 'GALAXY REST API' in cmd.stdout
    # Ensure we get data out of a specific collection
    cmd = host.run(_run_cmd(host, 4448, url='/api/v2/collections/community/general/'))
    assert 'https://{}:4448/api/'.format(hostname) in cmd.stdout
    answer = json.loads(cmd.stdout)
    version_uri = answer['latest_version']['href'].replace('https://{}:4448'.format(hostname), '')
    # Ensure we get a correct download URI
    cmd = host.run(_run_cmd(host, 4448, url=version_uri))
    assert 'https://{}:4448/api/'.format(hostname) in cmd.stdout
    answer = json.loads(cmd.stdout)
    download_uri = answer['download_url']
    assert download_uri.startswith('https://{}:4448/download/community-general'.format(hostname))
    # Download a file and check we get an actual archive
    download_uri = download_uri.replace('https://{}:4448'.format(hostname), '')
    host.run(_run_cmd(host, 4448, url=download_uri, curl_opt='-sL --output /tmp/output.tar.gz'))
    check_file = host.run('file /tmp/output.tar.gz')
    assert 'gzip compressed data' in check_file.stdout