quay/util/config/provider/testprovider.py

import json
import io

from util.config.provider.baseprovider import BaseProvider

REAL_FILES = ['test/data/signing-private.gpg', 'test/data/signing-public.gpg']

class TestLicense(object):
  @property
  def subscription(self):
    return {}

  @property
  def is_expired(self):
    return False

  def validate(self, config):
    pass

class TestConfigProvider(BaseProvider):
  """ Implementation of the config provider for testing. Everything is kept in-memory instead on
      the real file system. """
  def __init__(self):
    self.clear()

  def clear(self):
    self.files = {}
    self._config = {}

  @property
  def provider_id(self):
    return 'test'

  def update_app_config(self, app_config):
    self._config = app_config

  def get_config(self):
    if not 'config.yaml' in self.files:
      return None

    return json.loads(self.files.get('config.yaml', '{}'))

  def save_config(self, config_obj):
    self.files['config.yaml'] = json.dumps(config_obj)

  def config_exists(self):
    return 'config.yaml' in self.files

  def volume_exists(self):
    return True

  def volume_file_exists(self, filename):
    if filename in REAL_FILES:
      return True

    return filename in self.files

  def save_volume_file(self, filename, flask_file):
    self.files[filename] = ''

  def write_volume_file(self, filename, contents):
    self.files[filename] = contents

  def get_volume_file(self, filename, mode='r'):
    if filename in REAL_FILES:
      return open(filename, mode=mode)

    return io.BytesIO(self.files[filename])

  def requires_restart(self, app_config):
    return False

  def get_license(self):
    return TestLicense()

  def reset_for_test(self):
    self._config['SUPER_USERS'] = ['devtable']
    self.files = {}