quay/test/testlogs.py

import math
import logging

from random import SystemRandom
from loremipsum import get_sentence

from data.buildlogs import BuildLogs


logger = logging.getLogger(__name__)


class TestBuildLogs(BuildLogs):
  TEST_BUILD_ID = 'deadbeef-dead-beef-dead-beefdeadbeef'

  def __init__(self, redis_host):
    super(TestBuildLogs, self).__init__(redis_host)
    self.last_command = 0
    self.logs = [self._generate_command()]
    self.commands = [{
      'index': 0,
      'message': self.logs[0]['message'],
    }]
    self.request_counter = 0
    self._generate_logs()

  def _generate_command(self):
    self.last_command += 1
    return {
      'message': 'Step %s : %s' % (self.last_command, get_sentence()),
      'is_command': True,
    }

  def _generate_logs(self):
    rand = SystemRandom()
    num_logs = rand.randint(1, 500)
    for _ in range(num_logs):
      if rand.randint(1, 50) == 1:
        cmd = self._generate_command()
        self.commands.append({
          'message': cmd['message'],
          'index': len(self.logs),
        })
        self.logs.append(cmd)
      else:
        self.logs.append({
          'message': get_sentence(),
        })

  @staticmethod
  def generate_image_completion(rand_func):
    images = {}
    for image_id in range(rand_func.randint(1, 11)):
      total = int(math.pow(abs(rand_func.gauss(0, 1000)), 2))
      current = rand_func.randint(0, total)
      image_id = 'image_id_%s' % image_id
      images[image_id] = {
        'total': total,
        'current': current,
      }
    return images

  @staticmethod
  def generate_fake_status():
    response = {
      'id': 'deadbeef-dead-beef-dead-beefdeadbeef',
      'status': None,
    }

    random = SystemRandom()
    phases = {
      'waiting': {},
      'starting': {
        'total_commands': 7,
        'current_command': 0,
      },
      'initializing': {},
      'error': {},
      'complete': {},
      'building': {
        'total_commands': 7,
        'current_command': random.randint(1, 7),
      },
      'pushing': {
        'total_commands': 7,
        'current_command': 7,
        'push_completion': random.random(),
        'image_completion': TestBuildLogs.generate_image_completion(random),
      },
    }

    phase = random.choice(phases.keys())
    response['phase'] = phase
    response['status'] = (phases[phase])

    return response

  def get_log_entries(self, build_id, start_index, end_index):
    if build_id == self.TEST_BUILD_ID:
      self.request_counter += 1
      if self.request_counter % 10 == 0:
        self._generate_logs()
      logger.debug('Returning logs %s:%s', start_index, end_index)
      if end_index >= 0:
        end_index += 1
      return (len(self.logs), self.logs[start_index:end_index])
    else:
      return super(TestBuildLogs, self).get_log_entries(build_id, start_index,
                                                        end_index)

  def get_commands(self, build_id):
    if build_id == self.TEST_BUILD_ID:
      self.request_counter += 1
      if self.request_counter % 10 == 0:
        self._generate_logs()
      return self.commands
    else:
      return super(TestBuildLogs, self).get_commands(build_id)

  def get_last_command(self, build_id):
    if build_id == self.TEST_BUILD_ID:
      self.request_counter += 1
      if self.request_counter % 10 == 0:
        self._generate_logs()
      return self.commands[-1]
    else:
      return super(TestBuildLogs, self).get_last_command(build_id)

  def get_status(self, build_id):
    if build_id == self.TEST_BUILD_ID:
      self.request_counter += 1
      if self.request_counter % 10 == 0:
        self._generate_logs()
      return self.generate_fake_status()
    else:
      return super(TestBuildLogs, self).get_status(build_id)