Skip to content
Snippets Groups Projects
populate-depot.py 63.9 KiB
Newer Older
# Copyright © 2019-2021 Collabora Ltd.
#
# SPDX-License-Identifier: MIT
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

"""
Build the steam-container-runtime (aka SteamLinuxRuntime) depot, either
from just-built files or by downloading a previous build.
"""

import argparse
import json
import logging
import os
import subprocess
import tempfile
import urllib.request
from contextlib import suppress
from typing import (
    Any,
    Dict,
    List,
    Optional,
    Sequence,
from debian.deb822 import (
    Sources,
)


HERE = os.path.dirname(os.path.abspath(__file__))


logger = logging.getLogger('populate-depot')


DEFAULT_PRESSURE_VESSEL_URI = (
    'https://repo.steampowered.com/pressure-vessel/snapshots'
)

DEFAULT_IMAGES_URI = (
    'https://repo.steampowered.com/steamrt-images-SUITE/snapshots'
)


class InvocationError(Exception):
    pass


class PressureVesselRelease:
    def __init__(
        self,
        *,
        cache: str = '.cache',
        ssh_host: str = '',
        ssh_path: str = '',
        uri: str = DEFAULT_PRESSURE_VESSEL_URI,
        version: str = ''
    ) -> None:
        self.cache = cache
        self.pinned_version = None      # type: Optional[str]
        self.ssh_host = ssh_host
        self.ssh_path = ssh_path
        self.uri = uri
        self.version = version

    def get_uri(
        self,
        filename: str,
        version: Optional[str] = None,
    ) -> str:
        uri = self.uri
        v = version or self.pinned_version or self.version or 'latest'
        return f'{uri}/{v}/{filename}'

    def get_ssh_path(
        self,
        filename: str,
        version: Optional[str] = None,
    ) -> str:
        ssh_host = self.ssh_host
        ssh_path = self.ssh_path
        v = version or self.pinned_version or self.version or 'latest'

        if not ssh_host or not ssh_path:
            raise RuntimeError('ssh host/path not configured')

        return f'{ssh_path}/{v}/{filename}'

    def fetch(
        self,
        filename: str,
        opener: urllib.request.OpenerDirector,
        version: Optional[str] = None,
    ) -> str:
        dest = os.path.join(self.cache, filename)

        if self.ssh_host and self.ssh_path:
            path = self.get_ssh_path(filename)
            logger.info('Downloading %r...', path)
            subprocess.run([
                'rsync',
                '--archive',
                '--partial',
                '--progress',
                self.ssh_host + ':' + path,
                dest,
            ], check=True)
        else:
            uri = self.get_uri(filename)
            logger.info('Downloading %r...', uri)

            with opener.open(uri) as response:
                with open(dest + '.new', 'wb') as writer:
                    shutil.copyfileobj(response, writer)

                os.rename(dest + '.new', dest)

        return dest

    def pin_version(
        self,
        opener: urllib.request.OpenerDirector,
    ) -> str:
        pinned = self.pinned_version

        if pinned is None:
            if self.ssh_host and self.ssh_path:
                path = self.get_ssh_path(filename='VERSION.txt')
                logger.info('Determining version number from %r...', path)
                pinned = subprocess.run([
                    'ssh', self.ssh_host,
                    'cat {}'.format(shlex.quote(path)),
                ], stdout=subprocess.PIPE).stdout.decode('utf-8').strip()
            else:
                uri = self.get_uri(filename='VERSION.txt')
                logger.info('Determining version number from %r...', uri)
                with opener.open(uri) as response:
                    pinned = response.read().decode('utf-8').strip()

            self.pinned_version = pinned

        return pinned


class Runtime:
    def __init__(
        self,
        name,
        *,
        suite: str,

        architecture: str = 'amd64,i386',
        cache: str = '.cache',
        images_uri: str = DEFAULT_IMAGES_URI,
        path: Optional[str] = None,
        ssh_host: str = '',
        ssh_path: str = '',
        version: str = '',
    ) -> None:
        self.architecture = architecture
        self.cache = cache
        self.images_uri = images_uri
        self.name = name
        self.path = path
        self.suite = suite
        self.ssh_host = ssh_host
        self.ssh_path = ssh_path
Loading
Loading full blame...