Blob Blame Raw

import datetime

import db.types
import db.pool
import db.cache
import page.root
import page.error
import action.actions
import session
from repoproxy import RepoProxy
from repotypes.git import RepotypeGit


class Server:
  def __init__(self, config):
    self.repotypes = {
      'git': RepotypeGit() }
    
    self.config = self.apply_defaults(config)
    self.check_config(self.config)

    self.dbtypebytype = db.types.bytype
    self.dbtypebychar = db.types.bychar
    self.dbpool = db.pool.Pool(self)
    self.dbcache = db.cache.Cache(self)

    self.sessions = session.SessionManager(self)
    
    self.repoproxy = RepoProxy(self)
    
    self.pageroot      = page.root.RootPage()
    self.pageforbidden = page.error.ErrorPage('403', 'Forbidden')
    self.pagenotfound  = page.error.ErrorPage('404', 'Not Found')
    
    self.actions = action.actions.actions
    
    for key, repotype in self.repotypes.items():
      repotype.configure(self, self.config['repositories'][key])

  
  def get_defaults(self):
    return self.apply_defaults(dict())


  def apply_defaults(self, config):
    config_db = config.get('db', dict())
    config_db_pool = config_db.get('pool', dict())
    config_db_cache = config_db.get('cache', dict())
    
    config_users = config.get('users', dict())
    
    config_session = config.get('session', dict())
    
    config_repositories = config.get('repositories', dict())
    config_repositories_git = config_repositories.get('git', dict())
    
    urlprefix = self.fixpath( config.get('urlprefix', '') )
    
    return {
      'protocol'      : str(config.get('protocol', 'https://')),
      'domain'        : str(config.get('domain', '')),
      'name'          : str(config.get('name', 'Earthworm')),
      'debug'         : bool(config.get('debug')),

      'urlprefix'     : urlprefix,
      'urldataprefix' : str(config.get('urldataprefix', urlprefix + '/data')),
      
      'db': {
        'connection'  : dict(config_db.get('connection', dict())),
        'prefix'      : str(config_db.get('prefix', '')),
        'retrytime'   : float(config_db.get('retrytime', 0)),
        'pool': {
          'read'      : int(config_db_pool.get('read' , 10)),
          'write'     : int(config_db_pool.get('write', 10)),
        },
        'cache': {
          'maxcount'  : int(config_db_cache.get('maxcount', 10000)),
        },
      },
        
      'users': {
        'salt'        : str(config_users.get('salt', 'ndina82899nda90pn0al')),
        'selfcreate'  : bool(config_users.get('selfcreate', False)),
        'selfdelete'  : bool(config_users.get('selfdelete', False)),
        'showlist'    : bool(config_users.get('showlist', False)),
        'showprofile' : bool(config_users.get('showprofile', True)),
      },

      'session': {
        'time'        : float(config_session.get('time', 30*60))
      },

      'repositories': {
        'git': {
          'internalurl' : str(config_repositories_git.get('internalurl', '')),
          'path'        : str(config_repositories_git.get('path', '')),
          'command'     : str(config_repositories_git.get('command', '/usr/bin/git')),
          'environment' : config_repositories_git.get('environment', dict()),
        },
      },
    }

          
  def check_config(self, config):
    assert config['protocol'] == 'http://' \
        or config['protocol'] == 'https://'
    assert config['db']['retrytime'] >= 0
    assert config['db']['prefix'] == '' \
        or config['db']['prefix'].isidentifier()
    assert config['db']['pool']['read']  > 0
    assert config['db']['pool']['write'] > 0
    assert config['db']['cache']['maxcount'] > 0
    assert config['session']['time'] > 0
    
    for i in config['domain'].split('.'):
      assert i.isidentifier()

    config['urlprefix'] = self.fixpath(config['urlprefix'])
    config['urldataprefix'] = self.fixpath(config['urldataprefix'])

    for key, repotype in self.repotypes.items():
      repotype.check_config(self, config['repositories'][key])
   

  def fixpath(self, path, root = False):
    assert type(path) is str
    while path and path[-1] == '/':
      del path[-1]
    if path != '':
      assert path[0] == '/'
      for i in path[1:].split('/'):
        assert i.isalnum()
    if root:
      assert path and path[0] == '/'
    return path


  def fixenv(self, environment):
    assert type(environment) is dict
    for k, v in environment:
      assert type(k) is str and k.isidentifier()
      assert type(v) is str
    return environment