PK rIm fabfile/__init__.pyfrom fabfile.development import *
from fabfile.environment import *
# Make sure the virtualenv is activated.
setup_virtualenv()
PK rIWW W fabfile/development.py# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import unittest
from fabric import api
def test(*test_regexps):
"""Run tests.
:param test_regexps: A list of test regexps to include.
"""
# late import to ensure the virtual env is active
from sst import (
filters,
loaders,
)
os.environ['DJANGO_SETTINGS_MODULE'] = \
'u1testutils.selftests.django_project.settings'
loader = loaders.TestLoader()
suite = loader.suiteClass()
suite.addTests(loader.discoverTestsFromTree('u1testutils'))
suite.addTests(loader.discoverTestsFromTree('setup_vm'))
suite = filters.include_regexps(test_regexps, suite)
# List the tests as we run them
runner = unittest.TextTestRunner(verbosity=2)
res = runner.run(suite)
print 'Totals: ran({0}), skipped({1}), errors({2}), failures({3})'.format(
res.testsRun, len(res.skipped), len(res.errors), len(res.failures))
if not res.wasSuccessful():
api.abort('Tests failed.')
PK rI]z fabfile/environment.py# -*- coding: utf-8 -*-
# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import sys
from fabric.api import env, local
VIRTUALENV = '.env'
def bootstrap():
local('rm -rf %s' % VIRTUALENV)
local("find -name '*.pyc' -delete")
setup_virtualenv()
_install_dependencies()
def setup_virtualenv():
created = False
virtual_env = os.environ.get('VIRTUAL_ENV', None)
if virtual_env is None:
if not os.path.exists(VIRTUALENV):
_create_virtualenv()
created = True
virtual_env = VIRTUALENV
env.virtualenv = os.path.abspath(virtual_env)
_activate_virtualenv()
return created
def _create_virtualenv():
if not os.path.exists(VIRTUALENV):
virtualenv_bin_path = local('which virtualenv', capture=True)
virtualenv_version = local('{0} {1} --version'.format(
sys.executable, virtualenv_bin_path), capture=True)
# We only care about major.minor revision numbers
version_strings = virtualenv_version.split('.')[0:1]
virtualenv_version = [int(x) for x in version_strings]
args = '--distribute --clear'
if virtualenv_version < [1, 7]:
args += ' --no-site-packages'
local('{0} {1} {2} {3}'.format(sys.executable, virtualenv_bin_path,
args, VIRTUALENV), capture=False)
def _activate_virtualenv():
activate_this = os.path.abspath(
'{0}/bin/activate_this.py'.format(env.virtualenv))
execfile(activate_this, dict(__file__=activate_this))
# Work around fabric unconditionally removing first entry in sys.path by
# adding a dummy one. Agreed with Leo on IRC that the workaround is ok as
# the plan is to stop using fabric in the long run and the bug itself is a
# blocker. -- vila 2013-09-25
sys.path.insert(0, '/I-dont-exist-blow-me-away-I-dont-care')
def _install_dependencies():
# it's possible to get "ImportError: No module named setuptools"
# when using pip<1.4 to upgrade a package that depends on setuptools.
run_in_virtualenv_local('pip install -U setuptools', capture=False)
run_in_virtualenv_local(
'pip install -U -r requirements.txt', capture=False)
def run_in_virtualenv_local(command, capture=True):
prefix = ''
virtual_env = env.get('virtualenv', None)
if virtual_env:
prefix = '. {0}/bin/activate && '.format(virtual_env)
command = prefix + command
return local(command, capture=capture)
PK rI-eW u1testutils/logging.pyfrom __future__ import absolute_import
import logging
from unittest import TestCase
# Originally written by:
# - Guillermo Gonzalez
# - Facundo Batista
# - Natalia Bidart
class LogHandlerTestCase(TestCase):
"""A mixin that adds a memento loghandler for testing logging."""
class MementoHandler(logging.Handler):
"""A handler class which stores logging records in a list.
From http://nessita.pastebin.com/mgc85uQT
"""
def __init__(self, *args, **kwargs):
"""Create the instance, and add a records attribute."""
logging.Handler.__init__(self, *args, **kwargs)
self.records = []
def emit(self, record):
"""Just add the record to self.records."""
self.records.append(record)
def check(self, level, msg, check_traceback=False):
"""Check that something is logged."""
result = False
for rec in self.records:
if rec.levelname == level:
result = str(msg) in rec.getMessage()
if not result and check_traceback:
result = str(msg) in rec.exc_text
if result:
break
return result
def setUp(self):
"""Add the memento handler to the root logger."""
super(LogHandlerTestCase, self).setUp()
self.memento_handler = self.MementoHandler()
self.root_logger = logging.getLogger()
self.root_logger.addHandler(self.memento_handler)
def tearDown(self):
"""Remove the memento handler from the root logger."""
self.root_logger.removeHandler(self.memento_handler)
super(LogHandlerTestCase, self).tearDown()
def assertLogLevelContains(self, level, message, check_traceback=False):
check = self.memento_handler.check(
level, message, check_traceback=check_traceback)
msg = ('Expected logging message/s could not be found:\n%s\n'
'Current logging records are:\n%s')
expected = '\t%s: %s' % (level, message)
records = ['\t%s: %s' % (r.levelname, r.getMessage())
for r in self.memento_handler.records]
self.assertTrue(check, msg % (expected, '\n'.join(records)))
PK rIcC C u1testutils/mail.py# -*- coding: utf-8 -*-
# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see (1, 3):
csrf_middleware = 'django.middleware.csrf.CsrfViewMiddleware'
else:
csrf_middleware = 'django.contrib.csrf.middleware.CsrfMiddleware'
# make sure csrf middleware is enabled
self.old_MIDDLEWARE_CLASSES = settings.MIDDLEWARE_CLASSES
settings.MIDDLEWARE_CLASSES = [
'django.middleware.common.CommonMiddleware',
csrf_middleware,
'django.contrib.sessions.middleware.SessionMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
]
def tearDown(self):
settings.MIDDLEWARE_CLASSES = self.old_MIDDLEWARE_CLASSES
super(CsrfMiddlewareEnabledTestCase, self).tearDown()
def get_csrf_token(self, response):
# get csrf token
csrf_token_re = re.compile(r"name='csrfmiddlewaretoken' value='(.*)'")
match = re.search(csrf_token_re, response.content)
token = match.group(1)
return token
class NeverCacheTestCase(TestCase):
def is_cacheable(self, response):
result = True
if 'Expires'in response:
expires = response['Expires']
now = http_date(time.time())
result &= expires > now
if 'Cache-Control' in response:
cache_control = response['Cache-Control']
values = map(string.strip, cache_control.split(','))
for value in values:
if '=' in value:
k, v = value.split('=')
if k == 'max-age':
result &= int(v) > 0
break
return result
PK rIѤ u1testutils/sst/config.pyimport os
from sst.actions import set_base_url
DEFAULT_BASE_URL = 'http://localhost:8000'
def set_base_url_from_env(default_url=DEFAULT_BASE_URL):
"""Set the base URL for SST tests from the env or default."""
set_base_url(get_base_url_from_env(default_url))
def get_base_url_from_env(default_url=DEFAULT_BASE_URL):
base_url = os.environ.get('SST_BASE_URL', default_url)
return base_url.rstrip('/')
PK rIb8 u1testutils/sst/__init__.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import logging
import re
import urlparse
import sst.actions
from functools import wraps
from urllib2 import unquote
logger = logging.getLogger('User test')
def log_action(log_func):
"""Decorator to log the call of an action method."""
def middle(f):
@wraps(f)
def inner(instance, *args, **kwargs):
class_name = str(instance.__class__.__name__)
docstring = f.__doc__
if docstring:
docstring = docstring.split('\n')[0].strip()
else:
docstring = f.__name__
log_line = '%r: %r. Arguments %r. Keyword arguments: %r.'
log_func(log_line, class_name, docstring, args, kwargs)
return f(instance, *args, **kwargs)
return inner
return middle
class Page(object):
"""Base class for the page objects used in acceptance testing.
Instance variables:
title -- The title of the page.
url_path -- The path of the page.
is_url_path_regex -- If True, the url path will be considered as a regular
expression.
headings1 -- A list with the expected text of the h1 elements. If it's
empty, the h1 elements will not be checked.
headings2 -- A list with the expected text of the h2 elements. If it's
empty, the h2 elements will not be checked.
qa_anchor -- An string with the expected qa id
"""
title = None
url_path = None
is_url_path_regex = False
headings1 = []
headings2 = []
qa_anchor = ''
def __init__(self, open_page=False):
super(Page, self).__init__()
if open_page:
self._open_page()
self.assert_page_is_open()
@log_action(logging.info)
def _open_page(self):
"""Open the page."""
if self.is_url_path_regex:
raise ValueError(
"We can't open a page with a regular expression on the path.")
else:
assert self.url_path is not None
sst.actions.go_to(self.url_path)
return self
def assert_page_is_open(self):
"""Assert that the page is open and that no oops are displayed."""
try:
assert not self._is_oops_displayed(), \
'An oops error is displayed: {0}'.format(
self._get_oops_element().text)
self.assert_url_path()
# qa_anchor should take precedence
# since checking for text should be
# deprecated behaviour
if self.qa_anchor:
self.assert_qa_anchor()
else:
self.assert_title()
if self.headings1:
self.assert_headings1()
if self.headings2:
self.assert_headings2()
except AssertionError:
self._log_errors()
raise
def _is_oops_displayed(self):
try:
self._get_oops_element()
return True
except AssertionError:
return False
def _get_oops_element(self):
# TODO this works for U1. Does it work the same for pay and SSO?
oops_class = 'yui3-error-visible'
return sst.actions.get_element(css_class=oops_class)
def assert_title(self):
"""Assert the title of the page."""
sst.actions.assert_title(self.title)
def assert_url_path(self):
"""Assert the path of the page URL."""
if not self.is_url_path_regex:
sst.actions.assert_equal(
self._get_current_url_path(), self.url_path)
else:
self._assert_url_path_match()
def _assert_url_path_match(self):
# Make sure that there are no more characters at the end of the path.
url_path_regexp = self.url_path + '$'
current_url_path = self._get_current_url_path()
assert re.match(url_path_regexp, current_url_path), \
"The current URL path {0} doesn't match {1}".format(
current_url_path, url_path_regexp)
def _get_current_url_path(self):
current_url = sst.actions.get_current_url()
return unquote(urlparse.urlparse(current_url).path)
def assert_headings1(self):
"""Assert the h1 elements of the page."""
self._assert_elements_text('h1', self.headings1)
def _assert_elements_text(self, tag, expected_texts):
elements_text = self._get_elements_text(tag)
assert elements_text == expected_texts, \
'Expected elements texts: {0}\n' \
'Actual elements texts: {1}'.format(
', '.join(expected_texts), ', '.join(elements_text))
def _get_elements_text(self, tag=None, css_class=None):
return map(lambda x: x.text, sst.actions.get_elements(
tag=tag, css_class=css_class))
def assert_headings2(self):
"""Assert the h2 elements of the page."""
self._assert_elements_text('h2', self.headings2)
def assert_qa_anchor(self):
"""Assert the qa anchor."""
sst.actions.assert_element(
tag='html', **{'data-qa-id': self.qa_anchor})
def _log_errors(self):
if sst.actions.exists_element(css_class='error'):
logger.error(
', '.join(self._get_elements_text(css_class='error')))
class StringHTMLPage(Page):
"""Fake page for testing, with the source in a string attribute."""
def __init__(self, page_source, url_path, qa_anchor, open_page=False):
# Do not call the super __init__ because it will assert that the page
# is open. We will do that manually for the tests that require it.
self.url_path = url_path
self.page_source = page_source
self.qa_anchor = qa_anchor
self._make_temp_page()
if open_page:
self._open_page()
def _make_temp_page(self):
with open(self.url_path, 'w') as page_file:
page_file.write(self.page_source)
PK rI % u1testutils/sst/selftests/__init__.pyPK rI 0 u1testutils/sst/selftests/acceptance/__init__.pyPK rI'wT T 8 u1testutils/sst/selftests/acceptance/test_string_page.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import shutil
import tempfile
from sst import actions, cases
import u1testutils.sst
class StringPageTestCase(cases.SSTTestCase):
xserver_headless = True
base_url = 'file://'
page_source = (
"""
"""
)
def setUp(self):
super(StringPageTestCase, self).setUp()
self.temp_directory = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.temp_directory)
self.page_url_path = os.path.join(self.temp_directory, 'test')
def test_assert_string_page_is_opened(self):
page = u1testutils.sst.StringHTMLPage(
self.page_source, self.page_url_path, qa_anchor='test-qa-anchor',
open_page=True)
page.assert_page_is_open()
def test_string_page_not_opened(self):
page = u1testutils.sst.StringHTMLPage(
self.page_source, self.page_url_path, qa_anchor='test-qa-anchor')
# The page is not opened.
actions.fails(page.assert_page_is_open)
# But its file is created.
with open(self.page_url_path) as page_file:
page_source = page_file.read()
self.assertEqual(page_source, self.page_source)
PK rI3VQ 1 u1testutils/sst/selftests/acceptance/test_page.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import tempfile
import bs4
import mock
import sst.runtests
from testtools.matchers import Contains
import u1testutils.logging
import u1testutils.sst
class StringHTMLPage(u1testutils.sst.Page):
def __init__(self, page_source, title, headings1, headings2, url_path,
is_url_path_regex, qa_anchor=None):
self.page_source = page_source
self.title = title
self.headings1 = headings1
self.headings2 = headings2
page_file_name = None
if url_path:
self.url_path = url_path
else:
page_file_name = self._make_temp_page()
self.url_path = page_file_name
self.is_url_path_regex = is_url_path_regex
if qa_anchor:
self.qa_anchor = qa_anchor
try:
super(StringHTMLPage, self).__init__(open_page=True)
finally:
if page_file_name:
os.remove(page_file_name)
def _make_temp_page(self):
page_file = tempfile.NamedTemporaryFile(delete=False)
page_file.write(self.page_source)
page_file.close()
return page_file.name
class AssertPageTestCase(
sst.runtests.SSTTestCase, u1testutils.logging.LogHandlerTestCase):
page_source = (
"""
Test title
Test h1 1
Test h2 1
Test paragraph 1
Test h1 2
Test h2 2
Test paragraph 2
"""
)
base_url = 'file://'
xserver_headless = True
def setUp(self):
super(AssertPageTestCase, self).setUp()
self.page_kwargs = dict(
page_source=self.page_source,
title='Test title',
headings1=['Test h1 1', 'Test h1 2'],
headings2=['Test h2 1', 'Test h2 2'],
# If the url path is not specifiec, it will be set to the name of
# the temp file.
url_path=None,
is_url_path_regex=False
)
def test_corect_page_is_open(self):
StringHTMLPage(**self.page_kwargs)
# No error means the page was opened and asserted.
def test_wrong_title(self):
self.page_kwargs['title'] = 'Wrong title'
self.assertRaises(AssertionError, StringHTMLPage, **self.page_kwargs)
def test_wrong_url_path(self):
self.page_kwargs['url_path'] = 'Wrong path'
self.assertRaises(AssertionError, StringHTMLPage, **self.page_kwargs)
def test_wrong_headings1_text(self):
self.page_kwargs['headings1'] = ['Test h1 1', 'Wrong h1']
error = self.assertRaises(
AssertionError, StringHTMLPage, **self.page_kwargs)
self.assertEqual(
error.message,
'Expected elements texts: Test h1 1, Wrong h1\n'
'Actual elements texts: Test h1 1, Test h1 2')
def test_wrong_headings2_text(self):
self.page_kwargs['headings2'] = ['Test h2 1', 'Wrong h2']
error = self.assertRaises(
AssertionError, StringHTMLPage, **self.page_kwargs)
self.assertEqual(
error.message,
'Expected elements texts: Test h2 1, Wrong h2\n'
'Actual elements texts: Test h2 1, Test h2 2')
def test_assert_url_path_with_regexp(self):
page = StringHTMLPage(**self.page_kwargs)
page.url_path = '.+'
page.is_url_path_regex = True
page.assert_url_path()
def test_wrong_url_path_with_a_match(self):
page = StringHTMLPage(**self.page_kwargs)
page.url_path = '/test_path'
page.is_url_path_regex = True
with mock.patch('sst.actions.get_current_url') as mock_action:
mock_url = 'http://test_netloc/wrong/test_path/wrong'
mock_action.return_value = mock_url
self.assertRaises(AssertionError, page.assert_url_path)
def test_wrong_url_path_with_a_suffix(self):
page = StringHTMLPage(**self.page_kwargs)
page.url_path = '/test_path'
page.is_url_path_regex = True
with mock.patch('sst.actions.get_current_url') as mock_action:
mock_url = 'http://test_netloc/test_path/wrong'
mock_action.return_value = mock_url
self.assertRaises(AssertionError, page.assert_url_path)
def test_assert_url_path_with_query(self):
page = StringHTMLPage(**self.page_kwargs)
page.url_path = '/test_path'
with mock.patch('sst.actions.get_current_url') as mock_action:
mock_action.return_value = 'http://test_netloc/test_path?query'
page.assert_url_path()
def test_assert_page_with_visible_oops(self):
soup = bs4.BeautifulSoup(self.page_source)
oops_element = soup.new_tag('div')
oops_element['class'] = 'yui3-error-visible'
oops_element.string = 'Test oops'
soup.body.append(oops_element)
self.page_kwargs['page_source'] = str(soup)
error = self.assertRaises(
AssertionError, StringHTMLPage, **self.page_kwargs)
self.assertThat(error.message, Contains('Test oops'))
def test_assert_wrong_page_with_error(self):
soup = bs4.BeautifulSoup(self.page_source)
error_element = soup.new_tag('span')
error_element['class'] = 'error'
error_element.string = 'Test error'
soup.body.append(error_element)
self.page_kwargs['page_source'] = str(soup)
self.page_kwargs['title'] = 'Wrong title'
self.assertRaises(
AssertionError, StringHTMLPage, **self.page_kwargs)
self.assertLogLevelContains('ERROR', 'Test error')
def test_assert_correct_qa_anchor(self):
# Add the qa anchor to the page source.
soup = bs4.BeautifulSoup(self.page_source)
soup.html['data-qa-id'] = 'test_anchor'
self.page_kwargs['page_source'] = str(soup)
self.page_kwargs['qa_anchor'] = 'test_anchor'
# The title will not be asserted.
self.page_kwargs['title'] = 'Wrong title'
# The headings1 will not be asserted.
self.page_kwargs['headings1'] = ['Wrong h1']
# The headings 2 will not be asserted.
self.page_kwargs['headings2'] = ['Wrong h2']
# If the instantiation doesn't fail, it means we asserted it's the
# correct page only checking the anchor.
StringHTMLPage(**self.page_kwargs)
def test_assert_wrong_qa_anchor(self):
# Add the qa anchor to the page source.
soup = bs4.BeautifulSoup(self.page_source)
soup.html['data-qa-id'] = 'test_anchor'
self.page_kwargs['page_source'] = str(soup)
self.page_kwargs['qa_anchor'] = 'wrong_anchor'
self.assertRaises(AssertionError, StringHTMLPage, **self.page_kwargs)
def test_assert_qa_anchor_not_in_html_tag(self):
# Add the qa anchor to the page source.
soup = bs4.BeautifulSoup(self.page_source)
soup.html['data-qa-id'] = 'test_html_anchor'
div_element = soup.new_tag('div')
div_element['data-qa-id'] = 'test_div_anchor'
soup.body.append(div_element)
self.page_kwargs['page_source'] = str(soup)
self.page_kwargs['qa_anchor'] = 'test_div_anchor'
self.assertRaises(AssertionError, StringHTMLPage, **self.page_kwargs)
PK rI * u1testutils/sst/selftests/unit/__init__.pyPK rI-E[0 0 , u1testutils/sst/selftests/unit/test_pages.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import contextlib
import logging
import mock
import testtools
import u1testutils.logging
from u1testutils.sst import log_action, Page
class PageWithoutBrowser(Page):
"""Page double that allows us to test instantiation without a browser."""
def __init__(self, *args, **kwargs):
self.page_opened = False
self.url_path = '/test/path/'
super(PageWithoutBrowser, self).__init__(*args, **kwargs)
def _open_page(self):
self.page_opened = True
with mock.patch('sst.actions.go_to'):
super(PageWithoutBrowser, self)._open_page()
def assert_page_is_open(self):
assert self.page_opened
# do not call super()
class PageWithoutCheck(Page):
def __init__(self):
# We overwrite the parent constructor to do nothing, because on the
# parent it will try to assert that the page is open. Here we don't
# have a real page, but we can test the different page methods with
# mocks if we skip the check on init. We also have acceptance tests
# that use real pages for full coverage.
pass
class InitPageTestCase(testtools.TestCase):
def test_open_page_on_instantiation(self):
page = PageWithoutBrowser(open_page=True)
self.assertTrue(page.page_opened)
def test_assert_closed_page_on_instantiation(self):
self.assertRaises(AssertionError, PageWithoutBrowser, open_page=False)
def test_open_page_without_path(self):
page = PageWithoutBrowser(open_page=True)
page.url_path = None
self.assertRaises(AssertionError, page._open_page)
def test_open_page_with_path_regex(self):
page = PageWithoutBrowser(open_page=True)
page.url_path = '/test/.*'
page.is_url_path_regex = True
self.assertRaises(ValueError, page._open_page)
def test_open_page_goes_to_url_path(self):
page = PageWithoutCheck()
page.url_path = '/test/path'
with mock.patch('sst.actions.go_to') as mock_action:
returned_page = page._open_page()
mock_action.assert_called_with('/test/path')
self.assertEquals(page, returned_page)
def test_assert_page_is_open_without_qa_anchor(self):
# All the mocked methods have acceptance tests.
with contextlib.nested(
mock.patch.object(Page, '_is_oops_displayed', return_value=False),
mock.patch.object(Page, 'assert_title'),
mock.patch.object(Page, 'assert_url_path'),
) as mock_checks:
with mock.patch.object(Page, 'assert_qa_anchor') as mock_anchor:
PageWithoutCheck().assert_page_is_open()
for mock_check in mock_checks:
mock_check.assert_called_once_with()
self.assertFalse(mock_anchor.called)
def test_assert_page_is_open_with_anchor(self):
# All the mocked methods have acceptance tests.
with contextlib.nested(
mock.patch.object(Page, '_is_oops_displayed', return_value=False),
mock.patch.object(Page, 'assert_qa_anchor'),
mock.patch.object(Page, 'assert_url_path')
) as mock_checks:
with mock.patch.object(Page, 'assert_title') as mock_title:
page = PageWithoutCheck()
page.qa_anchor = 'test_anchor'
page.assert_page_is_open()
for mock_check in mock_checks:
mock_check.assert_called_once_with()
self.assertFalse(mock_title.called)
class PageWithOnlyHeadingsAssertions(PageWithoutCheck):
def assert_title(self):
pass
def assert_url_path(self):
pass
def _is_oops_displayed(self):
pass
class PageHeadingsTestCase(testtools.TestCase):
def test_assert_page_without_headings1_check(self):
with mock.patch.object(Page, 'assert_headings1') as mock_assert:
page = PageWithOnlyHeadingsAssertions()
page.headings1 = []
page.assert_page_is_open()
assert not mock_assert.called
def test_assert_page_without_headings2_check(self):
with mock.patch.object(Page, 'assert_headings2') as mock_assert:
page = PageWithOnlyHeadingsAssertions()
page.headings2 = []
page.assert_page_is_open()
assert not mock_assert.called
class PageWithLogDecorator(PageWithoutCheck):
@log_action(logging.info)
def do_something_without_docstring(self, *args, **kwargs):
pass
@log_action(logging.info)
def do_something_with_docstring(self, *args, **kwargs):
"""Do something with docstring."""
pass
@log_action(logging.info)
def do_something_with_multiline_docstring(self, *args, **kwargs):
"""Do something with a multiline docstring.
This should not be logged.
"""
pass
class PageLoggingTestCase(u1testutils.logging.LogHandlerTestCase):
def setUp(self):
super(PageLoggingTestCase, self).setUp()
self.root_logger.setLevel(logging.INFO)
self.page = PageWithLogDecorator()
def test_logged_action_without_docstring(self):
self.page.do_something_without_docstring(
'arg1', 'arg2', arg3='arg3', arg4='arg4')
self.assertLogLevelContains(
'INFO',
"'PageWithLogDecorator': 'do_something_without_docstring'. "
"Arguments ('arg1', 'arg2'). "
"Keyword arguments: {'arg3': 'arg3', 'arg4': 'arg4'}.")
def test_logged_action_with_docstring(self):
self.page.do_something_with_docstring(
'arg1', 'arg2', arg3='arg3', arg4='arg4')
self.assertLogLevelContains(
'INFO',
"'PageWithLogDecorator': 'Do something with docstring.'. "
"Arguments ('arg1', 'arg2'). "
"Keyword arguments: {'arg3': 'arg3', 'arg4': 'arg4'}.")
def test_logged_action_with_multiline_docstring(self):
self.page.do_something_with_multiline_docstring(
'arg1', 'arg2', arg3='arg3', arg4='arg4')
self.assertLogLevelContains(
'INFO',
"'PageWithLogDecorator': "
"'Do something with a multiline docstring.'. "
"Arguments ('arg1', 'arg2'). "
"Keyword arguments: {'arg3': 'arg3', 'arg4': 'arg4'}.")
PK rIn8uT T u1testutils/pay/api.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import payclient
from django.conf import settings
import u1testutils.pay.environment
from piston_mini_client import auth
class APIClient(object):
"""API client helper for payment tests.
It takes the credentials from the Django configuration file to make it
easier for the tests to call the API methods.
"""
def __init__(self, pay_server_url=None):
pay_server_url = self._get_pay_server_url(pay_server_url)
user = settings.PAY_API_USERNAME
password = settings.PAY_API_PASSWORD
basic_auth = auth.BasicAuthorizer(user, password)
pay_api_url = '{0}/api/2.0/'.format(pay_server_url)
self.client = payclient.PaymentServiceAPI(
auth=basic_auth, service_root=pay_api_url)
def _get_pay_server_url(self, pay_server_url=None):
if pay_server_url is None:
pay_server_url = u1testutils.pay.environment.get_pay_base_url()
else:
pay_server_url = pay_server_url.rstrip('/')
return pay_server_url
def add_new_credit_card(
self, shop_id, user, credit_card, billing_address,
make_unattended_default=False):
"""Add a new credit card through the Pay API.
Keyword arguments:
shop_id -- The identifier of the shop that will add the credit card.
user -- The user that will have the new credit card. It must have the
openid attribute.
credit_card -- The credit card information. It must have the attributes
card_type, name_on_cad, card_number, ccv_number, expiration_month and
expiration_year.
billing_address -- The billing address for the card. It must have the
attributes street_line1, state, country and postal_code. The
country is the name of the country in english.
make_unattended_default -- Boolean that indicates if the credit card
must be the new default method for unattended payments. Default is
False.
"""
credit_card_request = payclient.CreditCardRequest.from_data_objects(
shop_id, user.openid, credit_card, billing_address,
make_unattended_default)
return self.client.new_credit_card(data=credit_card_request)
PK rI@ u1testutils/pay/__init__.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import data
import environment
from api import APIClient
__all__ = [
'APIClient',
'data',
'environment',
]
PK rI[; ; u1testutils/pay/environment.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
from django.conf import settings
def get_pay_base_url(default='http://localhost:8000'):
base_url = getattr(settings, 'PAY_SERVER_URL', default).rstrip('/')
return base_url
PK rId0¤
u1testutils/pay/data.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see (1, 3):
included, excluded = excluded, included
self.testcase.setUp()
self.assertIn(included, settings.MIDDLEWARE_CLASSES)
self.assertNotIn(excluded, settings.MIDDLEWARE_CLASSES)
PK rIyH^ * u1testutils/selftests/unit/test_logging.pyfrom mock import Mock
from u1testutils.logging import LogHandlerTestCase
class LogHandlerTestCaseTestCase(LogHandlerTestCase):
def test_assertLogLevelContains_check_traceback(self):
mock_record = Mock()
mock_record.levelname = 'ERROR'
mock_record.getMessage.return_value = 'some message'
mock_record.exc_text = 'the exception'
self.memento_handler.emit(mock_record)
self.assertLogLevelContains('ERROR', 'some message',
check_traceback=True)
def test_MementoHandler_check_check_traceback(self):
mock_record = Mock()
mock_record.levelname = 'ERROR'
mock_record.getMessage.return_value = 'some message'
mock_record.exc_text = 'the exception'
self.memento_handler.emit(mock_record)
result = self.memento_handler.check('ERROR', 'some message',
check_traceback=True)
self.assertTrue(result)
PK rI ' u1testutils/selftests/unit/test_mail.py# -*- coding: utf-8 -*-
# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import u1testutils
from u1testutils.static import (
test_pep8_conformance,
test_pyflakes_analysis,
)
class Pep8ConformanceTestCase(
test_pep8_conformance.Pep8ConformanceTestCase):
packages = [u1testutils]
class PyflakesAnalysisTestCase(
test_pyflakes_analysis.PyflakesAnalysisTestCase):
packages = [u1testutils]
PK rI 0 u1testutils/selftests/django_project/__init__.pyPK rIX9 0 u1testutils/selftests/django_project/settings.py# -*- coding: utf-8 -*-
# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import quopri
import re
import u1testutils.mail
# The regex accepts 6-character tokens handed out by SSO between
# 2010 and 2015, and 20-character tokens with reduced charset handed
# out starting in November, 2015.
TOKEN_REGEX = re.compile(
r'/([A-Za-z0-9]{6}|'
'[bcdfghjkmnpqrtvwxyzBCDFGHJKLMNPQRTVWXYZ2346789]{20})/', re.S)
VALIDATION_LINK_PATTERNS = [
'/confirm-account/',
'/+newaccount/',
'/+newemail',
'/+resetpassword/',
]
INVALIDATION_LINK_PATTERNS = [
'/invalidate-email/',
]
def _get_verification_data_for_address(email_address):
"""A private helper for public helpers below.
Note: We have two different public helpers here for verification
code and link so that functional tests don't need to deal with
idioms like:
vcode, ignored = get_verification_for_address(email_address).
"""
email_msg = u1testutils.mail.get_latest_email_sent_to(email_address)
vcode = validation_link = invalidation_link = None
if email_msg:
# The body is encoded as quoted-printable. This affects any
# line longer than a certain length. Decode now to not have
# to worry about it in the regexen.
body = quopri.decodestring(email_msg.get_payload())
links = map(str.strip, re.findall('^(http.*)$', body, re.MULTILINE))
if len(links) > 0:
for link in links:
if any(pat in link for pat in VALIDATION_LINK_PATTERNS):
validation_link = link
match = TOKEN_REGEX.search(link)
if match:
vcode = match.group(1).strip()
elif any(pat in link for pat in INVALIDATION_LINK_PATTERNS):
invalidation_link = link
else:
match = re.search(
'Here is your confirmation code:(.*)(Enter|If you made)',
body, re.S)
if match:
vcode = match.group(1).strip()
if not vcode:
msg = "No verification code found in email. Email is:\n%r."
raise AssertionError(msg % body)
return vcode, validation_link, invalidation_link
def get_verification_code_for_address(email_address):
print("Retrieving verification code for %s." % email_address)
vcode, _, _ = _get_verification_data_for_address(email_address)
print("Verification code retrieved: %s." % vcode)
return vcode
def get_verification_link_for_address(email_address):
print("Retrieving verification link for %s." % email_address)
_, link, _ = _get_verification_data_for_address(email_address)
print("Verification link retrieved: %s." % link)
return link
def get_invalidation_link_for_address(email_address):
print("Retrieving invalidation link for %s." % email_address)
_, _, link = _get_verification_data_for_address(email_address)
print("Invalidation link retrieved: %s." % link)
return link
PK rIM g u1testutils/sso/api.py# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
from ssoclient.v2 import V2ApiClient
from u1testutils.sso import environment
class APIClient(object):
"""API client helper for identity tests."""
def __init__(self, sso_server_url=None):
sso_server_url = self._get_sso_server_url(sso_server_url)
self.client = V2ApiClient(sso_server_url + '/api/v2')
def _get_sso_server_url(self, sso_server_url=None):
if sso_server_url is None:
sso_server_url = environment.get_sso_base_url()
else:
sso_server_url = sso_server_url.rstrip('/')
return sso_server_url
def get_account_openid(self, email, password, token_name):
response = self.client.login(
email=email, password=password, token_name=token_name)
data = response.json()
openid = data.get('consumer_key')
return openid
def create_new_account(self, user, captcha_id=None, captcha_solution=None):
data = dict(
email=user.email,
password=user.password,
displayname=user.full_name
)
if captcha_id is not None:
data['captcha_id'] = captcha_id
if captcha_solution is not None:
data['captcha_solution'] = captcha_solution
response = self.client.register(data)
return response.status_code == 201
PK rI u1testutils/sso/__init__.pyPK rIyp p u1testutils/sso/environment.py# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
from __future__ import absolute_import
from django.conf import settings
def get_sso_base_url(default='http://localhost:8000'):
base_url = getattr(settings, 'OPENID_SSO_SERVER_URL', default).rstrip('/')
return base_url
PK rIZٌp^
^
u1testutils/sso/data.py# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import uuid
from django.conf import settings
from u1testutils.sso import api
class User(object):
def __init__(self, full_name, email, password, openid=None):
self.full_name = full_name
self.email = email
self.password = password
self._openid = openid
def __repr__(self):
return '%s(%r)' % (self.__class__, self.__dict__)
@property
def openid(self):
if self._openid is None:
token_name = "token-%s" % self.email
api_client = api.APIClient()
self._openid = api_client.get_account_openid(
email=self.email, password=self.password,
token_name=token_name)
return self._openid
@classmethod
def make_unique(cls, unique_id=None):
"""Return a unique user.
unique_id -- If a unique_id is passed, it will be used to uniquely
identify the data of the new user. If None is passed as unique_id,
then a uuid will be used. Default is None.
"""
if unique_id is None:
unique_id = str(uuid.uuid1())
full_name = 'Test user ' + unique_id
email = 'test+{0}@example.com'.format(unique_id)
password = 'Hola123*'
return cls(full_name, email, password)
@classmethod
def make_from_configuration(cls, new_user=False, unique_id=None):
"""Return a user taking its credentials from the configuration files.
Keyword arguments:
new_user -- If True, the full name and email of the user will have a
UUID to make them practically unique. In this case, the email will
be formed with the value of the EMAIL_ADDRESS_PATTERN
configuration variable. If False, the full name and email will be
the values of the SSO_TEST_ACCOUNT_FULL_NAME and
SSO_TEST_ACCOUNT_EMAIL configuration variables, respectively. In
both cases, the password will be the value of the
SSO_TEST_ACCOUNT_PASSWORD configuration variable.
Default is False.
unique_id -- If new_user is True and a unique_id is passed, then it
will be used to uniquely identify the data of the new user. If
None is passed as unique_id, then a uuid will be used. Default is
None.
"""
if new_user:
if unique_id is None:
unique_id = str(uuid.uuid1())
full_name = 'Test user ' + unique_id
email = settings.EMAIL_ADDRESS_PATTERN % unique_id
else:
full_name = settings.SSO_TEST_ACCOUNT_FULL_NAME
email = settings.SSO_TEST_ACCOUNT_EMAIL
password = settings.SSO_TEST_ACCOUNT_PASSWORD
return cls(full_name, email, password)
PK rI 4 4 u1testutils/sso/sst/pages.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import logging
import sst.actions
import u1testutils.sst
from u1testutils.sst import log_action
logger = logging.getLogger('User test')
class LogIn(u1testutils.sst.Page):
"""Log in page of the Ubuntu Single Sign On website.
This is a subclass of the PageWithAnonymousSubheader object to add
methods for the actions available in this page.
"""
url_path = '(/|/\+login)'
is_url_path_regex = True
qa_anchor = 'login'
@log_action(logging.info)
def _open_page(self):
"""Open the page.
This overwrites the method from the base Page class in order to choose
one of the possible URL paths to open the page.
"""
sst.actions.go_to('/+login')
return self
@log_action(logging.info)
def log_in_to_site_recognized(self, user=None):
"""Fill the log in form and continue to the site that requested it.
Keyword arguments:
user -- The user credentials. It must have the attributes email and
password. If None is passed as the user, it means that the user
has already started session on the identity provider and it's not
necessary to enter the credentials again.
"""
self._log_in(user)
@log_action(logging.info)
def log_in_to_site_not_recognized(self, user=None):
"""Fill the log in form and continue to the next step.
As the site is not recognized, the next step is the page where the
user can select the information that will be send to the site.
Keyword arguments:
user -- The user credentials. It must have the attributes email and
password. If None is passed as the user, it means that the user
has already started session on the identity provider and it's not
necessary to enter the credentials again.
"""
self._log_in(user)
return SiteNotRecognized()
def _log_in(self, user=None):
if user is not None:
sst.actions.wait_for(sst.actions.assert_title, 'Log in')
self._fill_log_in_form(user.email, user.password)
self._click_continue_button()
else:
# If None is passed as the user, it means that the user has
# already started session on the identity provider and it's not
# necessary to enter the credentials again.
pass
def _fill_log_in_form(self, email, password):
email_field = sst.actions.get_element_by_css(
'*[data-qa-id="login_form"] #id_email')
sst.actions.write_textfield(email_field, email)
password_field = sst.actions.get_element_by_css(
'*[data-qa-id="login_form"] #id_password')
sst.actions.write_textfield(password_field, password)
def _click_continue_button(self):
continue_button = sst.actions.get_element_by_css(
'*[data-qa-id="login_button"]')
sst.actions.click_button(continue_button)
@log_action(logging.info)
def go_to_create_new_account(self):
"""Go to the Create new account page."""
self._click_create_new_account()
return self._new_create_account_page()
def _click_create_new_account(self):
link = sst.actions.get_element_by_css(
'*[data-qa-id="create_account_link"]')
sst.actions.click_link(link)
def _new_create_account_page(self):
# override in children
return CreateAccount()
class LogInFromRedirect(LogIn):
url_path = '/.*/\+decide'
is_url_path_regex = True
@log_action(logging.info)
def go_to_create_new_account(self):
"""Go to the Create new account page."""
self._click_create_new_account()
return CreateAccountFromRedirect()
class PageWithAnonymousSubheader(u1testutils.sst.Page):
def __init__(self, open_page=False):
super(PageWithAnonymousSubheader, self).__init__(open_page)
self.subheader = _AnonymousSubheader()
class _AnonymousSubheader(object):
@log_action(logging.info)
def go_to_log_in_or_create_account(self):
link = sst.actions.get_element_by_css('*[data-qa-id="login_link"]')
sst.actions.click_link(link)
return LogIn()
class CreateAccount(PageWithAnonymousSubheader):
"""Create account page of the Ubuntu Single Sign On website.
This is a subclass of the PageWithAnonymousSubheader object to add
methods for the actions available in this page.
"""
url_path = '/+new_account'
qa_anchor = 'new_account'
@log_action(logging.info)
def create_ubuntu_sso_account(self, user):
"""Fill the new account form and continue to the next step.
Keyword arguments:
user -- The user credentials. It must have the attributes email and
password.
"""
self._fill_new_account_form(user)
self._click_continue()
def _fill_new_account_form(self, user, password_confirmation=None):
if password_confirmation is None:
password_confirmation = user.password
sst.actions.write_textfield('id_displayname', user.full_name)
email_field = sst.actions.get_element_by_css(
'*[data-qa-id="create_account_form"] #id_email')
sst.actions.write_textfield(email_field, user.email)
password_field = sst.actions.get_element_by_css(
'*[data-qa-id="create_account_form"] #id_password')
sst.actions.write_textfield(password_field, user.password)
sst.actions.write_textfield(
'id_passwordconfirm', password_confirmation)
captcha_field = 'recaptcha_response_field'
if sst.actions.exists_element(id=captcha_field):
sst.actions.write_textfield(captcha_field, 'ignored')
tos_field = 'id_accept_tos'
if sst.actions.exists_element(id=tos_field):
sst.actions.set_checkbox_value(tos_field, True)
def _click_continue(self):
continue_button = sst.actions.get_element_by_css(
'*[data-qa-id="register_button"]')
sst.actions.click_button(continue_button)
class CreateAccountFromRedirect(CreateAccount):
url_path = '/.*/\+new_account'
is_url_path_regex = True
class UnifiedLogInCreateAccount(LogIn, CreateAccount):
"""Log in or create account, unified experience."""
url_path = LogIn.url_path
is_url_path_regex = LogIn.is_url_path_regex
qa_anchor = 'login'
def _click_create_new_account(self):
new_user_intention = sst.actions.get_element_by_css(
'*[data-qa-id="user_intention_create"]')
sst.actions.click_element(new_user_intention)
def _new_create_account_page(self):
return self
class UnifiedLogInCreateAccountFromRedirect(UnifiedLogInCreateAccount):
url_path = '/.*/\+decide'
is_url_path_regex = True
class AccountCreationMailSent(PageWithAnonymousSubheader):
"""AccountCreation mail sent page of the Ubuntu Single Sign On website.
This is a subclass of the PageWithAnonymousSubheader object to add methods
for the actions available in this page.
"""
url_path = '/+new_account'
qa_anchor = 'new_account'
@log_action(logging.info)
def confirm_email_to_site_recognized(self, confirmation_code):
"""Confirm email and continue to the site that requested the log in.
Keyword arguments:
confirmation_code -- The confirmation code sent to the user email
address.
"""
self._confirm_email(confirmation_code)
@log_action(logging.info)
def confirm_email_to_site_not_recognized(self, confirmation_code):
"""Enter the confirmation code and continue to the next step.
As the site is not recognized, the next step is the page where the
user can select the information that will be send to the site.
Keyword arguments:
confirmation_code -- The confirmation code sent to the user email
address.
"""
self._confirm_email(confirmation_code)
return SiteNotRecognized()
def _confirm_email(self, confirmation_code):
self._enter_confirmation_code(confirmation_code)
self._click_continue_button()
def _enter_confirmation_code(self, confirmation_code):
confirmation_code_text_field = sst.actions.get_element(
name='confirmation_code')
sst.actions.write_textfield(confirmation_code_text_field,
confirmation_code)
def _click_continue_button(self):
continue_button = sst.actions.get_element_by_css(
'*[data-qa-id="complete_account_creation"]')
sst.actions.click_button(continue_button)
class PageWithUserSubheader(u1testutils.sst.Page):
def __init__(self, open_page=False):
# Some pages will need to take the user name from the subheader in
# order to assert that they are open. We need the subheader before
# that check.
self.subheader = UserSubheader()
super(PageWithUserSubheader, self).__init__(open_page)
class UserSubheader(object):
@log_action(logging.info)
def log_out(self):
"""Log out from the web site."""
link = sst.actions.get_element_by_css('*[data-qa-id="logout_link"]')
sst.actions.click_link(link)
return YouHaveBeenLoggedOut()
def get_user_name(self):
return sst.actions.get_element_by_css('*[data-qa-id="user_name"]').text
class CompleteEmailValidation(PageWithUserSubheader):
"""Complete email address validation page.
This is a subclass of the Page object. It adds methods for the actions
available in this page.
"""
url_path = '/token/.+/\+newemail/.+@.+'
is_url_path_regex = True
qa_anchor = 'confirm_email'
def __init__(self, open_page=False):
super(CompleteEmailValidation, self).__init__(open_page)
def _click_continue_button(self):
continue_button = sst.actions.get_element_by_css(
'*[data-qa-id="confirm_email_validation"]')
sst.actions.click_button(continue_button)
def _click_cancel(self):
cancel_link = sst.actions.get_element_by_css(
'*[data-qa-id="cancel_email_validation"]')
sst.actions.click_link(cancel_link)
@log_action(logging.info)
def confirm(self):
self._click_continue_button()
return YourAccount()
@log_action(logging.info)
def cancel(self):
return YourAccount()
class SiteNotRecognized(u1testutils.sst.Page):
"""Site not Recognized page of the Ubuntu Single Sign On website.
This is a subclass of the Page object. It overrides the assert_title
method to check only the first part of the title, and adds methods for the
actions available in this page.
"""
url_path = '/.+/\+decide'
is_url_path_regex = True
qa_anchor = 'decide'
def assert_title(self):
"""Assert that the page is open.
We use a regular expression because the title has the URL of the site
that requested the log in plus some tokens, and we don't need to check
that.
"""
sst.actions.assert_title_contains(self.title, regex=True)
@log_action(logging.info)
def make_all_information_available_to_website(self):
"""Select all the user available information.
This information will be send to the site that requested the log in.
"""
information_checkboxes = self._get_information_checkboxes()
for checkbox in information_checkboxes:
if checkbox.is_displayed():
sst.actions.set_checkbox_value(checkbox, True)
return self
def _get_information_checkboxes(self):
return sst.actions.wait_for(
sst.actions.get_elements_by_css,
'*[data-qa-id="info-items"] input[type="checkbox"]')
@log_action(logging.info)
def yes_sign_me_in(self):
"""Accept to sign in to the site not recognized and go back to it."""
sign_me_in_button = sst.actions.get_element_by_css(
'*[data-qa-id="rp_confirm_login"]')
sst.actions.click_button(sign_me_in_button)
class YourAccount(PageWithUserSubheader):
"""Your account page of the Ubuntu Single Sign On website.
This is a subclass of the Page object. It extends the constructor to
to receive the user name, and adds methods for the actions available in
this page.
Instance variables:
title -- The title of the page. It's build when the page is instantiated
using the user name.
"""
url_path = '/'
qa_anchor = 'edit_account'
def __init__(self, open_page=False):
super(YourAccount, self).__init__(open_page)
class YouHaveBeenLoggedOut(PageWithAnonymousSubheader):
"""Your account page of the Ubuntu Single Sign On website."""
url_path = '/+logout'
qa_anchor = 'logout'
PK rI W u1testutils/sso/sst/__init__.py# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import logging
import sst.actions
from u1testutils.sso import mail
from u1testutils.sso.sst import pages
logger = logging.getLogger('User test')
def create_new_account(user, is_site_recognized=True):
"""Create a new account on Ubuntu Single Sign On.
The browser must be on the Log in page.
If the account creation succeeds, the browser will be back on the site
that requested the log in.
Keyword arguments:
user -- The user object with the information for the new account.
It must have the full_name, email and password attributes.
is_site_recognized -- Boolean indicating if the site that requested the
log in is recognized. If it is not recognized, all the user available
information will be send to it. Default is True.
"""
log_in = pages.UnifiedLogInCreateAccountFromRedirect()
create_account = log_in.go_to_create_new_account()
create_account.create_ubuntu_sso_account(user)
if not is_site_recognized:
site_not_recognized = pages.SiteNotRecognized()
site_not_recognized.make_all_information_available_to_website()
site_not_recognized.yes_sign_me_in()
else:
_assert_site_not_recognized_page_not_opened()
def _assert_site_not_recognized_page_not_opened():
try:
sst.actions.fails(pages.SiteNotRecognized)
except AssertionError:
suggestion = (
'Please check that you are accessing SSO from a server that is '
'trusted. Otherwise, the unexpected Site Not Recognized page will '
'be opened.')
logger.error(suggestion)
raise
def validate_user_email(user):
"""Validate user's email address.
The user must be logged in and received the verification email
(for example, right after registration).
If the this succeeds, the browser will be redirected to user's
account page.
Keyword arguments:
user -- The user object with the information for the log in.
It must have the email and full_name attributes.
"""
vlink = mail.get_verification_link_for_address(user.email)
sst.actions.go_to(vlink)
validate_email = pages.CompleteEmailAddressValidation()
validate_email.confirm()
def sign_in(user=None, is_site_recognized=True):
"""Log in with an Ubuntu Single Sign On account.
The browser must be on the Log in page.
If the log in succeeds, the browser will be back on the site that
requested the log in.
Keyword arguments:
user -- The user object with the information for the log in.
It must have the email and password attributes.
If its value is None, it means that the user has already started a
session on Ubuntu Single Sign On and it's not necessary to enter the
credentials again. Default is None.
is_site_recognized -- Boolean indicating if the site that requested the
log in is recognized. If it is not recognized, all the user available
information will be send to it. Default is True.
"""
if is_site_recognized:
if user is not None:
log_in = pages.LogInFromRedirect()
log_in.log_in_to_site_recognized(user)
else:
# User is already signed in so SSO will just redirect back to the
# main site.
pass
_assert_site_not_recognized_page_not_opened()
else:
if user is not None:
log_in = pages.LogInFromRedirect()
site_not_recognized = log_in.log_in_to_site_not_recognized(user)
else:
# User is already signed in so SSO will just show the site not
# recognized page.
site_not_recognized = pages.SiteNotRecognized()
site_not_recognized.make_all_information_available_to_website()
site_not_recognized.yes_sign_me_in()
def log_out():
"""Log out from the Ubuntu Single Sign On site.
The browser must be on the Single Sign On site, and the user must have the
session started there.
If the log out succeeds, the browser will be on the Ubuntu Single Sign On
logout page.
"""
your_account = pages.YourAccount()
return your_account.subheader.log_out()
PK rI % u1testutils/sso/selftests/__init__.pyPK rI 0 u1testutils/sso/selftests/acceptance/__init__.pyPK rI\+ + 8 u1testutils/sso/selftests/acceptance/test_sst_helpers.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import shutil
import tempfile
import uuid
import testscenarios
from sst import cases
import u1testutils.sst
from u1testutils import logging
from u1testutils.sso import (
data,
sst as sso_sst,
)
_WRONG_PAGE_SOURCE = (
"""
"""
)
_SITE_NOT_RECOGNIZED_PAGE_SOURCE = (
"""
"""
)
_CREATE_ACCOUNT_FROM_REDIRECT_PAGE_SOURCE = (
"""
"""
)
_UNIFIED_LOGIN_FROM_REDIRECT_PAGE_SOURCE = (
"""
Log in
create account
"""
)
_UNIFIED_CREATE_ACCOUNT_FROM_REDIRECT_PAGE_SOURCE = (
"""
Log in
"""
)
class FakeSSOWebsite(object):
def __init__(self, is_user_logged_in, is_site_recognized, brand):
super(FakeSSOWebsite, self).__init__()
self.is_user_logged_in = is_user_logged_in
self.is_site_recognized = is_site_recognized
self.brand = brand
def __enter__(self):
self.temp_directory = tempfile.mkdtemp()
self._write_fake_pages()
def _write_fake_pages(self):
if self.is_site_recognized:
not_recognized_path = ''
else:
not_recognized_path = self._write_fake_site_not_recognized_page()
if not self.is_user_logged_in:
self._write_login_and_create_account_pages(not_recognized_path)
def _write_fake_site_not_recognized_page(self):
source = _SITE_NOT_RECOGNIZED_PAGE_SOURCE
path = '+decide'
qa_anchor = 'decide'
return self._write_fake_page_with_token(path, source, qa_anchor)
def _write_fake_page_with_token(self, path, source, qa_anchor, **links):
# We do not check the tokens, so the easiest thing is to have the
# every page in separate paths, to write them to HTML files and be able
# to open all of them on the same test. In real life, the tokens are
# the same. To check the token can be a future improvement.
# -- elopio - 2013-07-03
token = str(uuid.uuid1())
token_directory = os.path.join(self.temp_directory, token)
os.mkdir(token_directory)
return self._write_fake_page(
token_directory, path, source, qa_anchor, **links)
def _write_fake_page(self, directory, path, source, qa_anchor, **links):
if links:
source = source.format(**links)
page_path = os.path.join(directory, path)
u1testutils.sst.StringHTMLPage(
source, page_path, qa_anchor, open_page=True)
return page_path
def _write_login_and_create_account_pages(self, next_page_path):
create_account_path = self._write_fake_unified_create_account_page(
next_page_path)
self._write_fake_unified_login_page(
next_page_path, create_account_path)
def _write_fake_unified_create_account_page(self, next_page_path):
source = _UNIFIED_CREATE_ACCOUNT_FROM_REDIRECT_PAGE_SOURCE
path = '+decide'
qa_anchor = 'login'
return self._write_fake_page_with_token(
path, source, qa_anchor,
create_account_form_submit_link=next_page_path)
def _write_fake_unified_login_page(
self, next_page_path, create_account_path):
source = _UNIFIED_LOGIN_FROM_REDIRECT_PAGE_SOURCE
path = '+decide'
qa_anchor = 'login'
return self._write_fake_page_with_token(
path, source, qa_anchor,
login_form_submit_link=next_page_path,
create_account_link=create_account_path)
def _write_fake_create_account_page(self, next_page_path):
source = _CREATE_ACCOUNT_FROM_REDIRECT_PAGE_SOURCE
path = '+new_account'
qa_anchor = 'new_account'
return self._write_fake_page_with_token(
path, source, qa_anchor,
create_account_form_submit_link=next_page_path)
def __exit__(self, exc_type, exc_val, exc_tb):
shutil.rmtree(self.temp_directory)
class WrongSSOSiteTestCase(cases.SSTTestCase):
xserver_headless = True
def setUp(self):
self.base_url = 'file://'
super(WrongSSOSiteTestCase, self).setUp()
self._write_wrong_page()
def _write_wrong_page(self):
temp_directory = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, temp_directory)
page_path = os.path.join(temp_directory, 'wrong')
u1testutils.sst.StringHTMLPage(
_WRONG_PAGE_SOURCE, page_path, qa_anchor='wrong',
open_page=True)
def test_sign_in(self):
self.assertRaises(
AssertionError, sso_sst.sign_in, user='not important',
is_site_recognized=False)
def test_create_account(self):
self.assertRaises(
AssertionError, sso_sst.sign_in, user='not important',
is_site_recognized=False)
class SignInTestCase(
testscenarios.TestWithScenarios, cases.SSTTestCase,
logging.LogHandlerTestCase):
scenarios = [
('ubuntuone brand', {'brand': 'ubuntuone'}),
('other brand', {'brand': 'other'}),
]
xserver_headless = True
def setUp(self):
self.base_url = 'file://'
super(SignInTestCase, self).setUp()
def test_sign_in_with_already_signed_user_to_site_recognized(self):
# Sign in will do nothing.
sso_sst.sign_in(user=None, is_site_recognized=True)
def test_sign_in_with_already_signed_user_to_site_not_recognized(self):
with FakeSSOWebsite(
is_user_logged_in=True, is_site_recognized=False,
brand=self.brand):
sso_sst.sign_in(user=None, is_site_recognized=False)
def test_sign_in_to_site_recognized(self):
user = data.User.make_unique()
with FakeSSOWebsite(
is_user_logged_in=False, is_site_recognized=True,
brand=self.brand):
sso_sst.sign_in(user, is_site_recognized=True)
def test_sign_in_to_site_not_recognized(self):
user = data.User.make_unique()
with FakeSSOWebsite(
is_user_logged_in=False, is_site_recognized=False,
brand=self.brand):
sso_sst.sign_in(user, is_site_recognized=False)
def test_sign_in_with_unexpected_site_not_recognized(self):
user = data.User.make_unique()
with FakeSSOWebsite(
is_user_logged_in=False, is_site_recognized=False,
brand=self.brand):
self.assertRaises(
AssertionError, sso_sst.sign_in, user, is_site_recognized=True)
self.assertLogLevelContains(
'ERROR',
'Please check that you are accessing SSO from a server that is '
'trusted. Otherwise, the unexpected Site Not Recognized page will '
'be opened.'
)
class CreateNewAccountTestCase(
testscenarios.TestWithScenarios, cases.SSTTestCase,
logging.LogHandlerTestCase):
scenarios = [
('ubuntuone brand', {'brand': 'ubuntuone'}),
('other brand', {'brand': 'other'}),
]
xserver_headless = True
def setUp(self):
self.temp_directory = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.temp_directory)
self.base_url = 'file://' + self.temp_directory
super(CreateNewAccountTestCase, self).setUp()
def test_create_account_to_site_recognized(self):
user = data.User.make_unique()
with FakeSSOWebsite(
is_user_logged_in=False, is_site_recognized=True,
brand=self.brand):
sso_sst.create_new_account(user, is_site_recognized=True)
def test_create_account_to_site_not_recognized(self):
user = data.User.make_unique()
with FakeSSOWebsite(
is_user_logged_in=False, is_site_recognized=False,
brand=self.brand):
sso_sst.create_new_account(user, is_site_recognized=False)
def test_create_account_with_unexpected_site_not_recognized(self):
user = data.User.make_unique()
with FakeSSOWebsite(
is_user_logged_in=False, is_site_recognized=False,
brand=self.brand):
self.assertRaises(
AssertionError, sso_sst.create_new_account, user,
is_site_recognized=True)
self.assertLogLevelContains(
'ERROR',
'Please check that you are accessing SSO from a server that is '
'trusted. Otherwise, the unexpected Site Not Recognized page will '
'be opened.'
)
PK rI_ / u1testutils/sso/selftests/unit/test_sso_mail.py# -*- coding: utf-8 -*-
# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see
To: to@example.com
Hello
As a final step of the Ubuntu Single Sign On (SSO) account creation process=
, please validate the email address to@example.com. Ubuntu SSO enables conv=
enient access to a variety of Ubuntu-related services like Ubuntu One with =
the same username and password.
Here is your confirmation code:
%(confirm)s
Enter this code into the registration form, or click the following link to =
automatically confirm your account:
https://login.ubuntu.com/confirm-account/%(confirm)s/to@example.com
If you don\'t know what this is about, then someone has probably entered you=
r email address by mistake. Sorry about that.
If you wish to report this email being incorrectly used, please click the f=
ollowing link:
https://login.ubuntu.com/invalidate-email/%(inval)s/to@example.com
You can also seek further assistance on:
https://forms.canonical.com/sso-support/
Thank you,
The Ubuntu Single Sign On team
https://login.ubuntu.com/
"""
class SSOMailTestCase(unittest.TestCase):
verification_email = email.parser.Parser().parsestr(
EMAIL_TEMPLATE % {'confirm': 'QM9CSR', 'inval': '42FcJG'}
)
verification_email_long_token = email.parser.Parser().parsestr(
EMAIL_TEMPLATE % {'confirm': 'dgfjL4kYvTgnCGkBh32n',
'inval': 'y74Yh6jkT6Wj89jRgDYy'}
)
urls = [
r'/%s/+new_account',
r'/invalidate-email/%s/a@b.com',
r'/confirm-account/%s/a@b.com',
r'/token/%s/+resetpassword/a@b.com',
r'/token/%s/+newemail/a@b.com',
]
def setUp(self):
super(SSOMailTestCase, self).setUp()
patcher = mock.patch('u1testutils.mail')
self.mock_mail = patcher.start()
self.addCleanup(patcher.stop)
def test_get_verification_code(self):
self.mock_mail.get_latest_email_sent_to.return_value = \
self.verification_email
verification_code = mail.get_verification_code_for_address(
'to@example.com')
self.assertEquals(verification_code, 'QM9CSR')
def test_get_verification_code_long_token(self):
self.mock_mail.get_latest_email_sent_to.return_value = \
self.verification_email_long_token
verification_code = mail.get_verification_code_for_address(
'to@example.com')
self.assertEquals(verification_code, 'dgfjL4kYvTgnCGkBh32n')
def test_get_verification_link(self):
self.mock_mail.get_latest_email_sent_to.return_value = \
self.verification_email
verification_link = mail.get_verification_link_for_address(
'to@example.com')
self.assertEquals(
verification_link,
'https://login.ubuntu.com/confirm-account/QM9CSR/to@example.com')
def test_get_verification_link_long_token(self):
self.mock_mail.get_latest_email_sent_to.return_value = \
self.verification_email_long_token
verification_link = mail.get_verification_link_for_address(
'to@example.com')
self.assertEquals(
verification_link,
'https://login.ubuntu.com/confirm-account/'
'dgfjL4kYvTgnCGkBh32n/to@example.com')
def test_get_invalidation_link(self):
self.mock_mail.get_latest_email_sent_to.return_value = \
self.verification_email
invalidation_link = mail.get_invalidation_link_for_address(
'to@example.com')
self.assertEquals(
invalidation_link,
'https://login.ubuntu.com/invalidate-email/42FcJG/to@example.com')
def test_get_invalidation_link_long_token(self):
self.mock_mail.get_latest_email_sent_to.return_value = \
self.verification_email_long_token
invalidation_link = mail.get_invalidation_link_for_address(
'to@example.com')
self.assertEquals(
invalidation_link,
'https://login.ubuntu.com/invalidate-email/'
'y74Yh6jkT6Wj89jRgDYy/to@example.com')
def test_no_code(self):
mock_email = mock.MagicMock()
mock_email.get_payload.return_value = 'XXXXXX'
self.mock_mail.get_latest_email_sent_to.return_value = mock_email
with self.assertRaises(AssertionError):
mail.get_verification_code_for_address('to@example.com')
def test_token_regex(self):
token = 'T0Kenz'
for pattern in self.urls:
for host in ('http://localhost:8000', 'https://login.ubuntu.com'):
url = host + pattern % token
m = mail.TOKEN_REGEX.search(url)
self.assertEqual(m.group(1), token)
def test_token_regex_long(self):
token = 'dgfjL4kYvTgnCGkBh32n'
for pattern in self.urls:
for host in ('http://localhost:8000', 'https://login.ubuntu.com'):
url = host + pattern % token
m = mail.TOKEN_REGEX.search(url)
self.assertEqual(m.group(1), token)
def test_token_regex_long_bad(self):
# 0 is not a valid long-token character
token = 'dgfjL4kYvTgnCGk0h32n'
for pattern in self.urls:
for host in ('http://localhost:8000', 'https://login.ubuntu.com'):
url = host + pattern % token
m = mail.TOKEN_REGEX.search(url)
self.assertIsNone(m)
PK rI6G* + u1testutils/sso/selftests/unit/test_data.py# -*- coding: utf-8 -*-
# Copyright 2012, 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import unittest
from django.conf import settings
from u1testutils.sso import environment
class EnvironmentTestCase(unittest.TestCase):
def test_get_sso_base_url_from_settings(self):
settings.OPENID_SSO_SERVER_URL = 'http://foo'
try:
base_url = environment.get_sso_base_url()
self.assertEqual(base_url, 'http://foo')
finally:
del settings.OPENID_SSO_SERVER_URL
def test_get_sso_base_url_from_settings_with_stripping(self):
settings.OPENID_SSO_SERVER_URL = 'http://foo/'
try:
base_url = environment.get_sso_base_url()
self.assertEqual(base_url, 'http://foo')
finally:
del settings.OPENID_SSO_SERVER_URL
def test_get_sso_base_url_from_default(self):
base_url = environment.get_sso_base_url()
self.assertEqual(base_url, 'http://localhost:8000')
PK rI0n" " , u1testutils/static/test_pyflakes_analysis.py# -*- coding: utf-8 -*-
# Copyright 2012 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import unittest
import cStringIO
from mock import patch
from pyflakes.scripts.pyflakes import checkPath
class PyflakesAnalysisTestCase(unittest.TestCase):
packages = []
exclude_file = None
def test_pyflakes_analysis(self):
self.assertNotEqual([], self.packages,
'You should define some packages to check')
string_out = cStringIO.StringIO()
with patch('sys.stdout', string_out):
for package in self.packages:
self._run_pyflakes_analysis(package)
results = string_out.getvalue().splitlines()
unexpected_errors = self._get_unexpected_errors(results)
self.assertEqual(unexpected_errors, [],
'\n'.join(unexpected_errors))
def _run_pyflakes_analysis(self, package):
package_path = os.path.dirname(package.__file__)
for dir_path, dir_names, file_names in os.walk(package_path):
for file_name in file_names:
if file_name.endswith('.py'):
file_path = os.path.join(dir_path, file_name)
checkPath(file_path)
def _get_unexpected_errors(self, results):
unexpected_errors = []
expected_errors = self._parse_exclude_file()
if not expected_errors:
unexpected_errors = results
else:
for line in results:
found = False
for file_, error in expected_errors:
if file_ in line and error in line:
found = True
break
if not found:
unexpected_errors.append(line)
return unexpected_errors
def _parse_exclude_file(self):
excluded_errors = []
if self.exclude_file is not None:
with open(self.exclude_file) as file_:
for line in file_:
if line.strip():
source_file, error = [data.strip() for data in
line.split(':', 1)]
excluded_errors.append((source_file, error))
return excluded_errors
PK rI=
"V V u1testutils/static/__init__.py# No tests to load here
from sst import loaders
discover = loaders.discoverNoTests
PK rI^ + u1testutils/static/test_pep8_conformance.py# -*- coding: utf-8 -*-
# Copyright 2012 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import os
import pep8
import unittest
from collections import defaultdict
class UnitTestReport(pep8.BaseReport):
def __init__(self, options):
super(UnitTestReport, self).__init__(options)
self._fmt = pep8.REPORT_FORMAT.get(options.format.lower(),
options.format)
self._msgs = []
def error(self, line_number, offset, text, check):
"""Report an error, according to options."""
super(UnitTestReport, self).error(line_number, offset,
text, check)
self._msgs.append(self._fmt % {
'path': self.filename,
'row': self.line_offset + line_number, 'col': offset + 1,
'code': text[:4], 'text': text[5:],
})
class Pep8ConformanceTestCase(unittest.TestCase):
packages = []
exclude = []
def setUp(self):
super(Pep8ConformanceTestCase, self).setUp()
self.pep8style = pep8.StyleGuide(
counters=defaultdict(int),
doctest='',
exclude=self.exclude,
filename=['*.py'],
ignore=[],
repeat=True,
select=[],
show_pep8=False,
show_source=True,
max_line_length=79,
quiet=0,
statistics=False,
testsuite='',
verbose=0
)
self.report = self.pep8style.init_report(UnitTestReport)
def test_pep8_conformance(self):
self.assertNotEqual([], self.packages,
'You should define some packages to check')
for package in self.packages:
self.pep8style.input_dir(os.path.dirname(package.__file__))
self.assertEqual(self.pep8style.options.report.total_errors, 0,
'\n'.join(self.report._msgs))
PK rIo6I setup_vm/__init__.py# Not a package, only needed to load tests.
import os
def discover(test_loader, package, directory, names):
return test_loader.discoverTestsFromTree(os.path.join(directory, 'tests'))
PK rIV ' setup_vm/tests/test_code_conformance.py# Copyright 2013 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
import setup_vm
from u1testutils.static import (
test_pep8_conformance as pep8,
test_pyflakes_analysis as pyflakes,
)
class Pep8ConformanceTestCase(pep8.Pep8ConformanceTestCase):
packages = [setup_vm]
class PyflakesAnalysisTestCase(pyflakes.PyflakesAnalysisTestCase):
packages = [setup_vm]
PK rIS^S S setup_vm/tests/test_setup_vm.pyfrom cStringIO import StringIO
import os
from bzrlib import errors
import testtools
from setup_vm import tests
from setup_vm.bin import setup_vm as svm
def requires_known_reference_image(test):
# We need a pre-seeded download cache from the user running the tests
# as downloading the cloud image is too long.
user_conf = svm.VmStack(None)
download_cache = user_conf.get('vm.download_cache')
if download_cache is None:
test.skip('vm.download_cache is not set')
# We use some known reference
reference_cloud_image_name = 'raring-server-cloudimg-amd64-disk1.img'
cloud_image_path = os.path.join(
download_cache, reference_cloud_image_name)
if not os.path.exists(cloud_image_path):
test.skip('%s is not available' % (cloud_image_path,))
return download_cache, reference_cloud_image_name
class TestCaseWithHome(testtools.TestCase):
"""Provide an isolated disk-based environment.
A $HOME directory is setup as well as an /etc/ one so tests can setup
config files.
"""
def setUp(self):
super(TestCaseWithHome, self).setUp()
tests.set_cwd_to_tmp(self)
tests.isolate_env(self)
# Isolate tests from the user environment
self.home_dir = os.path.join(self.test_base_dir, 'home')
os.mkdir(self.home_dir)
os.environ['HOME'] = self.home_dir
# Also isolate from the system environment
self.etc_dir = os.path.join(self.test_base_dir, 'etc')
os.mkdir(self.etc_dir)
self.patch(svm, 'system_config_dir', lambda: self.etc_dir)
class TestVmMatcher(TestCaseWithHome):
def setUp(self):
super(TestVmMatcher, self).setUp()
self.store = svm.VmStore('.', 'foo.conf')
self.matcher = svm.VmMatcher(self.store, 'test')
def test_empty_section_always_matches(self):
self.store._load_from_string('foo=bar')
matching = list(self.matcher.get_sections())
self.assertEqual(1, len(matching))
def test_specific_before_generic(self):
self.store._load_from_string('foo=bar\n[test]\nfoo=baz')
matching = list(self.matcher.get_sections())
self.assertEqual(2, len(matching))
# First matching section is for test
self.assertEqual(self.store, matching[0][0])
base_section = matching[0][1]
self.assertEqual('test', base_section.id)
# Second matching section is the no-name one
self.assertEqual(self.store, matching[0][0])
no_name_section = matching[1][1]
self.assertIs(None, no_name_section.id)
class TestVmStores(TestCaseWithHome):
def setUp(self):
super(TestVmStores, self).setUp()
self.conf = svm.VmStack('foo')
def test_default_in_empty_stack(self):
self.assertEqual('1024', self.conf.get('vm.ram_size'))
def test_system_overrides_internal(self):
self.conf.system_store._load_from_string('vm.ram_size = 42')
self.assertEqual('42', self.conf.get('vm.ram_size'))
def test_user_overrides_system(self):
self.conf.system_store._load_from_string('vm.ram_size = 42')
self.conf.store._load_from_string('vm.ram_size = 4201')
self.assertEqual('4201', self.conf.get('vm.ram_size'))
def test_local_overrides_user(self):
self.conf.system_store._load_from_string('vm.ram_size = 42')
self.conf.store._load_from_string('vm.ram_size = 4201')
self.conf.local_store._load_from_string('vm.ram_size = 8402')
self.assertEqual('8402', self.conf.get('vm.ram_size'))
class TestVmStack(TestCaseWithHome):
"""Test config option values."""
def setUp(self):
super(TestVmStack, self).setUp()
self.conf = svm.VmStack('foo')
self.conf.store._load_from_string('''
vm.release=raring
vm.cpu_model=amd64
''')
def assertValue(self, expected_value, option):
self.assertEqual(expected_value, self.conf.get(option))
def test_raring_iso_url(self):
self.assertValue('http://cdimage.ubuntu.com/daily-live/current/',
'vm.iso_url')
def test_raring_iso_name(self):
self.assertValue('raring-desktop-amd64.iso', 'vm.iso_name')
def test_raring_cloud_image_url(self):
self.assertValue('http://cloud-images.ubuntu.com/raring/current/',
'vm.cloud_image_url')
def test_raring_cloud_image_name(self):
self.assertValue('raring-server-cloudimg-amd64-disk1.img',
'vm.cloud_image_name')
def test_apt_proxy_set(self):
proxy = 'apt_proxy: http://example.org:4321'
self.conf.set('vm.apt_proxy', proxy)
self.assertEqual(proxy, self.conf.expand_options('{vm.apt_proxy}'))
def test_download_cache_with_user_expansion(self):
download_cache = '~/installers'
self.conf.set('vm.download_cache', download_cache)
self.assertValue(os.path.join(self.home_dir, 'installers'),
'vm.download_cache')
def test_images_dir_with_user_expansion(self):
images_dir = '~/images'
self.conf.set('vm.images_dir', images_dir)
self.assertValue(os.path.join(self.home_dir, 'images'),
'vm.images_dir')
class TestPathOption(TestCaseWithHome):
def assertConverted(self, expected, value):
option = svm.PathOption('foo', help='A path.')
self.assertEquals(expected, option.convert_from_unicode(None, value))
def test_absolute_path(self):
self.assertConverted('/test/path', '/test/path')
def test_home_path_with_expansion(self):
self.assertConverted(self.home_dir, '~')
def test_path_in_home_with_expansion(self):
self.assertConverted(os.path.join(self.home_dir, 'test/path'),
'~/test/path')
class TestDownload(TestCaseWithHome):
# FIXME: Needs parametrization agains
# vm.{cloud_image_name|cloud_tarball_name|iso_name} an
# {download_iso|download_cloud_image|download_cloud_tarball} {Lxc|Kvm}...
# Maybe we just need to test _download_in_cache() now that it's implemented
# at V level and be done -- vila 2013-08-06
def setUp(self):
tests.requires_feature(self, tests.use_sudo_for_tests_feature)
# Downloading real isos or images is too long for tests, instead, we
# fake it by downloading a small but known to exist file: MD5SUMS
super(TestDownload, self).setUp()
download_cache = os.path.join(self.test_base_dir, 'downloads')
os.mkdir(download_cache)
self.conf = svm.VmStack('foo')
self.conf.store._load_from_string('''
vm.iso_name=MD5SUMS
vm.cloud_image_name=MD5SUMS
vm.cloud_tarball_name=MD5SUMS
vm.release=raring
vm.cpu_model=amd64
vm.download_cache=%s
''' % (download_cache,))
def test_download_iso(self):
vm = svm.Kvm(self.conf)
self.assertTrue(vm.download_iso())
# Trying to download again will find the file in the cache
self.assertFalse(vm.download_iso())
# Forcing the download even when the file is present
self.assertTrue(vm.download_iso(force=True))
def test_download_unknown_iso_fail(self):
self.conf.set('vm.iso_name', 'I-dont-exist')
vm = svm.Kvm(self.conf)
self.assertRaises(svm.CommandError, vm.download_iso)
def test_download_iso_with_unknown_cache_fail(self):
dl_cache = os.path.join(self.test_base_dir, 'I-dont-exist')
self.conf.set('vm.download_cache', dl_cache)
vm = svm.Kvm(self.conf)
self.assertRaises(svm.ConfigValueError, vm.download_iso)
def test_download_cloud_image(self):
vm = svm.Kvm(self.conf)
self.assertTrue(vm.download_cloud_image())
# Trying to download again will find the file in the cache
self.assertFalse(vm.download_cloud_image())
# Forcing the download even when the file is present
self.assertTrue(vm.download_cloud_image(force=True))
def test_download_unknown_cloud_image_fail(self):
self.conf.set('vm.cloud_image_name', 'I-dont-exist')
vm = svm.Kvm(self.conf)
self.assertRaises(svm.CommandError, vm.download_cloud_image)
def test_download_cloud_image_with_unknown_cache_fail(self):
dl_cache = os.path.join(self.test_base_dir, 'I-dont-exist')
self.conf.set('vm.download_cache', dl_cache)
vm = svm.Kvm(self.conf)
self.assertRaises(svm.ConfigValueError, vm.download_cloud_image)
class TestMetaData(TestCaseWithHome):
def setUp(self):
super(TestMetaData, self).setUp()
self.conf = svm.VmStack('foo')
self.vm = svm.Kvm(self.conf)
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir)
config_dir = os.path.join(self.test_base_dir, 'config')
self.conf.store._load_from_string('''
vm.name=foo
vm.images_dir=%s
vm.config_dir=%s
''' % (images_dir, config_dir,))
def test_create_meta_data(self):
self.vm.create_meta_data()
self.assertTrue(os.path.exists(self.vm._config_dir))
self.assertTrue(os.path.exists(self.vm._meta_data_path))
with open(self.vm._meta_data_path) as f:
meta_data = f.readlines()
self.assertEqual(2, len(meta_data))
self.assertEqual('instance-id: foo\n', meta_data[0])
self.assertEqual('local-hostname: foo\n', meta_data[1])
class TestYaml(testtools.TestCase):
def yaml_load(self, *args, **kwargs):
return svm.yaml.safe_load(*args, **kwargs)
def yaml_dump(self, *args, **kwargs):
return svm.yaml.safe_dump(*args, **kwargs)
def test_load_scalar(self):
self.assertEqual(
{'foo': 'bar'}, self.yaml_load(StringIO('{foo: bar}')))
# Surprisingly the enclosing braces are not needed, probably a special
# case for the highest level
self.assertEqual({'foo': 'bar'}, self.yaml_load(StringIO('foo: bar')))
def test_dump_scalar(self):
self.assertEqual('{foo: bar}\n', self.yaml_dump(dict(foo='bar')))
def test_load_list(self):
self.assertEqual({'foo': ['a', 'b', 'c']},
# Single space indentation is enough
self.yaml_load(StringIO('''\
foo:
- a
- b
- c
''')))
def test_dump_list(self):
# No more enclosing braces... yeah for consistency :-/
self.assertEqual(
'foo: [a, b, c]\n', self.yaml_dump(dict(foo=['a', 'b', 'c'])))
def test_load_dict(self):
self.assertEqual({'foo': {'bar': 'baz'}},
self.yaml_load(StringIO('{foo: {bar: baz}}')))
multiple_lines = '''\
foo: {bar: multiple
lines}
'''
self.assertEqual(
{'foo': {'bar': 'multiple lines'}},
self.yaml_load(StringIO(multiple_lines)))
class TestLaunchpadAccess(TestCaseWithHome):
def setUp(self):
super(TestLaunchpadAccess, self).setUp()
self.conf = svm.VmStack('foo')
self.vm = svm.Kvm(self.conf)
self.ci_data = svm.CIUserData(self.conf)
def test_cant_find_private_key(self):
self.conf.store._load_from_string('vm.launchpad_id = I-dont-exist')
e = self.assertRaises(svm.ConfigPathNotFound,
self.ci_data.set_launchpad_access)
key_path = '~/.ssh/I-dont-exist@setup_vm'
self.assertEqual(key_path, e.path)
msg_prefix = 'You need to create the {p} keypair'.format(p=key_path)
self.assertTrue(unicode(e).startswith(msg_prefix))
def test_id_with_key(self):
ssh_dir = os.path.join(self.home_dir, '.ssh')
os.mkdir(ssh_dir)
key_path = os.path.join(ssh_dir, 'user@setup_vm')
with open(key_path, 'w') as f:
f.write('key content')
self.conf.store._load_from_string('vm.launchpad_id = user')
self.assertEqual(None, self.ci_data.launchpad_hook)
self.ci_data.set_launchpad_access()
self.assertEqual('''\
#!/bin/sh
mkdir -p /home/ubuntu/.ssh
chown ubuntu:ubuntu ~ubuntu
chmod 0700 ~ubuntu
chown ubuntu:ubuntu /home/ubuntu/.ssh
chmod 0700 /home/ubuntu/.ssh
cat >/home/ubuntu/.ssh/id_rsa <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
key content
EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown ubuntu:ubuntu /home/ubuntu/.ssh/id_rsa
chmod 0400 /home/ubuntu/.ssh/id_rsa
''',
self.ci_data.launchpad_hook)
cc = self.ci_data.cloud_config
self.assertEquals([['sudo', '-u', 'ubuntu',
'bzr', 'launchpad-login', 'user']],
cc['runcmd'])
class TestCIUserData(TestCaseWithHome):
def setUp(self):
super(TestCIUserData, self).setUp()
self.conf = svm.VmStack('foo')
self.ci_data = svm.CIUserData(self.conf)
def test_empty_config(self):
self.ci_data.populate()
# Check default values
self.assertIs(None, self.ci_data.root_hook)
self.assertIs(None, self.ci_data.ubuntu_hook)
cc = self.ci_data.cloud_config
self.assertFalse(cc['apt_update'])
self.assertFalse(cc['apt_upgrade'])
self.assertEqual({'expire': False}, cc['chpasswd'])
self.assertEqual('setup_vm finished installing in ${uptime} seconds.',
cc['final_message'])
self.assertTrue(cc['manage_etc_hosts'])
self.assertEqual('ubuntu', cc['password'])
self.assertEqual({'mode': 'poweroff'}, cc['power_state'])
def test_password(self):
self.conf.store._load_from_string('vm.password = tagada')
self.ci_data.populate()
self.assertEquals('tagada', self.ci_data.cloud_config['password'])
def test_apt_proxy(self):
self.conf.store._load_from_string('vm.apt_proxy = tagada')
self.ci_data.populate()
self.assertEquals('tagada', self.ci_data.cloud_config['apt_proxy'])
def test_final_message_precise(self):
self.conf.store._load_from_string('vm.release = precise')
self.ci_data.populate()
self.assertEqual('setup_vm finished installing in $UPTIME seconds.',
self.ci_data.cloud_config['final_message'])
def test_poweroff_precise(self):
self.conf.store._load_from_string('vm.release = precise')
self.ci_data.populate()
self.assertEqual(['halt'], self.ci_data.cloud_config['runcmd'])
def test_poweroff_quantal(self):
self.conf.store._load_from_string('vm.release = quantal')
self.ci_data.populate()
self.assertEqual(['halt'], self.ci_data.cloud_config['runcmd'])
def test_poweroff_other(self):
self.conf.store._load_from_string('vm.release = raring')
self.ci_data.populate()
self.assertEqual(
{'mode': 'poweroff'}, self.ci_data.cloud_config['power_state'])
self.assertIs(None, self.ci_data.cloud_config.get('runcmd'))
def test_update_true(self):
self.conf.store._load_from_string('vm.update = True')
self.ci_data.populate()
cc = self.ci_data.cloud_config
self.assertTrue(cc['apt_update'])
self.assertTrue(cc['apt_upgrade'])
def test_packages(self):
self.conf.store._load_from_string('vm.packages = bzr,ubuntu-desktop')
self.ci_data.populate()
self.assertEqual(['bzr', 'ubuntu-desktop'],
self.ci_data.cloud_config['packages'])
def test_apt_sources(self):
self.conf.store._load_from_string('''\
vm.release = raring
# Ensure options are properly expanded (and comments supported ;)
_archive_url = http://archive.ubuntu.com/ubuntu
_ppa_url = https://u:p@ppa.lp.net/user/ppa/ubuntu
vm.apt_sources = deb {_archive_url} {vm.release} partner,\
deb {_archive_url} {vm.release} main,\
deb {_ppa_url} {vm.release} main|ABCDEF
''')
self.ci_data.populate()
self.assertEqual(
[{'source': 'deb http://archive.ubuntu.com/ubuntu raring partner'},
{'source': 'deb http://archive.ubuntu.com/ubuntu raring main'},
{'source':
'deb https://u:p@ppa.lp.net/user/ppa/ubuntu raring main',
'keyid': 'ABCDEF'}],
self.ci_data.cloud_config['apt_sources'])
def create_file(self, path, content):
with open(path, 'wb') as f:
f.write(content)
def test_good_ssh_keys(self):
paths = ('rsa', 'rsa.pub', 'dsa', 'dsa.pub', 'ecdsa', 'ecdsa.pub')
for path in paths:
self.create_file(path, '%s\ncontent\n' % (path,))
paths_as_list = ','.join(paths)
self.conf.store._load_from_string(
'vm.ssh_keys = %s' % (paths_as_list,))
self.ci_data.populate()
self.assertEqual({'dsa_private': 'dsa\ncontent\n',
'dsa_public': 'dsa.pub\ncontent\n',
'ecdsa_private': 'ecdsa\ncontent\n',
'ecdsa_public': 'ecdsa.pub\ncontent\n',
'rsa_private': 'rsa\ncontent\n',
'rsa_public': 'rsa.pub\ncontent\n'},
self.ci_data.cloud_config['ssh_keys'])
def test_bad_type_ssh_keys(self):
self.conf.store._load_from_string('vm.ssh_keys = I-dont-exist')
self.assertRaises(svm.ConfigValueError, self.ci_data.populate)
def test_unknown_ssh_keys(self):
self.conf.store._load_from_string('vm.ssh_keys = rsa.pub')
self.assertRaises(svm.ConfigPathNotFound, self.ci_data.populate)
def test_good_ssh_authorized_keys(self):
paths = ('home.pub', 'work.pub')
for path in paths:
self.create_file(path, '%s\ncontent\n' % (path,))
paths_as_list = ','.join(paths)
self.conf.store._load_from_string(
'vm.ssh_authorized_keys = %s' % (paths_as_list,))
self.ci_data.populate()
self.assertEqual(['home.pub\ncontent\n', 'work.pub\ncontent\n'],
self.ci_data.cloud_config['ssh_authorized_keys'])
def test_unknown_ssh_authorized_keys(self):
self.conf.store._load_from_string('vm.ssh_authorized_keys = rsa.pub')
self.assertRaises(svm.ConfigPathNotFound, self.ci_data.populate)
def test_unknown_root_script(self):
self.conf.store._load_from_string('vm.root_script = I-dont-exist')
self.assertRaises(svm.ConfigPathNotFound, self.ci_data.populate)
def test_unknown_ubuntu_script(self):
self.conf.store._load_from_string('vm.ubuntu_script = I-dont-exist')
self.assertRaises(svm.ConfigPathNotFound, self.ci_data.populate)
def test_expansion_error_in_script(self):
root_script_content = '''#!/bin/sh
echo Hello {I_dont_exist}
'''
with open('root_script.sh', 'w') as f:
f.write(root_script_content)
self.conf.store._load_from_string('''\
vm.root_script = root_script.sh
''')
e = self.assertRaises(errors.ExpandingUnknownOption,
self.ci_data.populate)
self.assertEqual(root_script_content, e.string)
def test_unknown_uploaded_scripts(self):
self.conf.store._load_from_string(
'''vm.uploaded_scripts = I-dont-exist ''')
self.assertRaises(svm.ConfigPathNotFound,
self.ci_data.populate)
def test_root_script(self):
with open('root_script.sh', 'w') as f:
f.write('''#!/bin/sh
echo Hello {user}
''')
self.conf.store._load_from_string('''\
vm.root_script = root_script.sh
user=root
''')
self.ci_data.populate()
# The additional newline after the script is expected
self.assertEqual('''\
#!/bin/sh
cat >~root/setup_vm_post_install <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
#!/bin/sh
echo Hello root
EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown root:root ~root/setup_vm_post_install
chmod 0700 ~root/setup_vm_post_install
''', self.ci_data.root_hook)
self.assertEqual(['~root/setup_vm_post_install'],
self.ci_data.cloud_config['runcmd'])
def test_ubuntu_script(self):
with open('ubuntu_script.sh', 'w') as f:
f.write('''#!/bin/sh
echo Hello {user}
''')
self.conf.store._load_from_string('''\
vm.ubuntu_script = ubuntu_script.sh
user = ubuntu
''')
self.ci_data.populate()
# The additional newline after the script is expected
self.assertEqual('''\
#!/bin/sh
cat >~ubuntu/setup_vm_post_install <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
#!/bin/sh
echo Hello ubuntu
EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown ubuntu:ubuntu ~ubuntu/setup_vm_post_install
chmod 0700 ~ubuntu/setup_vm_post_install
''', self.ci_data.ubuntu_hook)
# The command is run as root, so we need to 'su ubuntu' first
self.assertEqual([['su',
'-l',
'-c',
'~ubuntu/setup_vm_post_install',
'ubuntu']],
self.ci_data.cloud_config['runcmd'])
def test_uploaded_scripts(self):
paths = ('foo', 'bar')
for path in paths:
self.create_file(path, '%s\ncontent\n' % (path,))
paths_as_list = ','.join(paths)
self.conf.store._load_from_string(
'vm.uploaded_scripts = %s' % (paths_as_list,))
self.ci_data.populate()
self.assertEqual('''\
#!/bin/sh
cat >~ubuntu/setup_vm_uploads <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
mkdir -p ~ubuntu/bin
cd ~ubuntu/bin
cat >foo <<'EOFfoo'
foo
content
EOFfoo
chmod 0755 foo
cat >bar <<'EOFbar'
bar
content
EOFbar
chmod 0755 bar
EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown ubuntu:ubuntu ~ubuntu/setup_vm_uploads
chmod 0700 ~ubuntu/setup_vm_uploads
''',
self.ci_data.uploaded_scripts_hook)
self.assertEqual([['su',
'-l',
'-c',
'~ubuntu/setup_vm_uploads',
'ubuntu']],
self.ci_data.cloud_config['runcmd'])
class TestCreateUserData(TestCaseWithHome):
def setUp(self):
super(TestCreateUserData, self).setUp()
self.conf = svm.VmStack('foo')
self.vm = svm.Kvm(self.conf)
def test_empty_config(self):
config_dir = os.path.join(self.test_base_dir, 'config')
os.mkdir(config_dir)
# The config is *almost* empty, we need to set config_dir though as the
# user-data needs to be stored there.
self.conf.store._load_from_string('vm.config_dir=%s' % (config_dir,))
self.vm.create_user_data()
self.assertTrue(os.path.exists(self.vm._config_dir))
self.assertTrue(os.path.exists(self.vm._user_data_path))
with open(self.vm._user_data_path) as f:
user_data = f.readlines()
# We care about the two first lines only here, checking the format (or
# cloud-init is confused)
self.assertEqual('#cloud-config-archive\n', user_data[0])
self.assertEqual("- {content: '#cloud-config\n", user_data[1])
class TestSeedData(TestCaseWithHome):
def setUp(self):
super(TestSeedData, self).setUp()
self.conf = svm.VmStack('foo')
self.vm = svm.VM(self.conf)
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir)
config_dir = os.path.join(self.test_base_dir, 'config')
self.conf.store._load_from_string('''
vm.name=foo
vm.release=raring
vm.config_dir=%s
vm.images_dir=%s
''' % (config_dir, images_dir,))
def test_create_meta_data(self):
self.vm.create_meta_data()
self.assertTrue(os.path.exists(self.vm._meta_data_path))
def test_create_user_data(self):
self.vm.create_user_data()
self.assertTrue(os.path.exists(self.vm._user_data_path))
class TestSeedImage(TestCaseWithHome):
def setUp(self):
tests.requires_feature(self, tests.use_sudo_for_tests_feature)
tests.requires_feature(self, tests.qemu_img_feature)
super(TestSeedImage, self).setUp()
self.conf = svm.VmStack('foo')
self.vm = svm.Kvm(self.conf)
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir)
config_dir = os.path.join(self.test_base_dir, 'config')
self.conf.store._load_from_string('''
vm.name=foo
vm.release=raring
vm.config_dir=%s
vm.images_dir=%s
''' % (config_dir, images_dir,))
def test_create_seed_image(self):
self.assertTrue(self.vm._seed_path is None)
self.vm.create_seed_image()
self.assertFalse(self.vm._seed_path is None)
self.assertTrue(os.path.exists(self.vm._seed_path))
class TestImageFromCloud(TestCaseWithHome):
def setUp(self):
tests.requires_feature(self, tests.use_sudo_for_tests_feature)
tests.requires_feature(self, tests.qemu_img_feature)
super(TestImageFromCloud, self).setUp()
self.conf = svm.VmStack('foo')
self.vm = svm.Kvm(self.conf)
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir)
download_cache_dir = os.path.join(self.test_base_dir, 'download')
os.mkdir(download_cache_dir)
self.conf.store._load_from_string('''
vm.name=foo
vm.release=raring
vm.images_dir=%s
vm.download_cache=%s
vm.cloud_image_name=fake.img
vm.disk_size=1M
''' % (images_dir, download_cache_dir))
def test_create_disk_image(self):
cloud_image_path = os.path.join(self.conf.get('vm.download_cache'),
self.conf.get('vm.cloud_image_name'))
# We need a fake cloud image that can be converted
svm.run_subprocess(
['sudo', 'qemu-img', 'create',
cloud_image_path, self.conf.get('vm.disk_size')])
self.assertTrue(self.vm._disk_image_path is None)
self.vm.create_disk_image()
self.assertFalse(self.vm._disk_image_path is None)
self.assertTrue(os.path.exists(self.vm._disk_image_path))
class TestImageWithBacking(TestCaseWithHome):
def setUp(self):
tests.requires_feature(self, tests.use_sudo_for_tests_feature)
(download_cache_dir,
reference_cloud_image_name) = requires_known_reference_image(self)
super(TestImageWithBacking, self).setUp()
# We'll share the images_dir between vms
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir)
# Create a shared config
conf = svm.VmStack(None)
conf.store._load_from_string('''
vm.release=raring
vm.images_dir=%s
vm.download_cache=%s
vm.disk_size=2G
[selftest-from-cloud]
vm.name=selftest-from-cloud
vm.cloud_image_name=%s
[selftest-backing]
vm.name=selftest-backing
vm.backing=selftest-from-cloud.qcow2
''' % (images_dir, download_cache_dir, reference_cloud_image_name))
conf.store.save()
# To bypass creating a real vm, we start from the cloud image that is a
# real and bootable one, so we just convert it. That also makes it
# available in vm.images_dir
temp_vm = svm.Kvm(svm.VmStack('selftest-from-cloud'))
temp_vm.create_disk_image()
def test_create_image_with_backing(self):
vm = svm.Kvm(svm.VmStack('selftest-backing'))
self.assertTrue(vm._disk_image_path is None)
vm.create_disk_image()
self.assertFalse(vm._disk_image_path is None)
self.assertTrue(os.path.exists(vm._disk_image_path))
class TestKvmStates(testtools.TestCase):
def assertStates(self, expected, lines):
self.assertEqual(expected, svm.kvm_states(lines))
def test_empty(self):
self.assertStates({}, [])
def test_garbage(self):
self.assertRaises(ValueError, self.assertStates, None, [''])
def test_known_states(self):
# From a real life sample
self.assertStates({'foo': 'shut off', 'bar': 'running'},
['- foo shut off',
'19 bar running'])
class TestLxcInfo(testtools.TestCase):
def assertInfo(self, expected, lines):
self.assertEqual(expected, svm.lxc_info('foo', lines))
def test_empty(self):
self.assertRaises(ValueError,
self.assertInfo, dict(state='STOPPED', pid=-1), [])
def test_garbage(self):
self.assertRaises(ValueError, self.assertInfo, None, [''])
def test_stopped(self):
# From a real life sample
self.assertInfo({'state': 'STOPPED', 'pid': '-1'},
['state: STOPPED',
'pid: -1'])
def test_running(self):
# From a real life sample
self.assertInfo({'state': 'RUNNING', 'pid': '30937'},
['state: RUNNING',
'pid: 30937'])
class TestConsoleParsing(testtools.TestCase):
def _parse_console_monitor(self, string):
mon = svm.ConsoleMonitor(StringIO(string))
lines = []
for line in mon.parse():
lines.append(line)
return lines
def test_fails_on_empty(self):
self.assertRaises(svm.ConsoleEOFError,
self._parse_console_monitor, '')
def test_fail_on_knwon_cloud_init_errors(self):
self.assertRaises(
svm.CloudInitError,
self._parse_console_monitor,
'Failed loading yaml blob\n')
self.assertRaises(
svm.CloudInitError,
self._parse_console_monitor,
'Unhandled non-multipart userdata starting\n')
self.assertRaises(
svm.CloudInitError,
self._parse_console_monitor,
"failed to render string to stdout: cannot find 'uptime'\n")
self.assertRaises(
svm.CloudInitError,
self._parse_console_monitor,
"Failed loading of cloud config "
"'/var/lib/cloud/instance/cloud-config.txt'. "
"Continuing with empty config\n")
def test_succeds_on_final_message(self):
lines = self._parse_console_monitor('''
Lalala
I'm doing my work
It goes nicely
setup_vm finished installing in 1 seconds.
That was fast isn't it ?
* Will now halt
[ 33.204755] Power down.
''')
# We stop as soon as we get the final message and ignore the rest
self.assertEquals(' * Will now halt\n',
lines[-1])
class TestConsoleParsingWithFile(TestCaseWithHome):
def _parse_file_monitor(self, string):
with open('console', 'w') as f:
f.write(string)
mon = svm.FileMonitor('console')
for line in mon.parse():
pass
return mon.lines
def test_succeeds_with_file(self):
content = '''\
Yet another install
Going well
setup_vm finished installing in 0.5 seconds.
Wow, even faster !
* Will now halt
Whatever, won't read that
'''
lines = self._parse_file_monitor(content)
expected_lines = content.splitlines(True)
# Remove the last line that should not be seen
expected_lines = expected_lines[:-1]
self.assertEqual(expected_lines, lines)
def xtest_fails_on_empty_file(self):
# FIXME: We need some sort of timeout there...
self.assertRaises(svm.CommandError, self._parse_file_monitor, '')
def test_fail_on_knwon_cloud_init_errors_with_file(self):
self.assertRaises(
svm.CloudInitError,
self._parse_file_monitor,
'Failed loading yaml blob\n')
self.assertRaises(
svm.CloudInitError,
self._parse_file_monitor,
'Unhandled non-multipart userdata starting\n')
self.assertRaises(
svm.CloudInitError,
self._parse_file_monitor,
"failed to render string to stdout: cannot find 'uptime'\n")
class TestInstallWithSeed(TestCaseWithHome):
def setUp(self):
tests.requires_feature(self, tests.use_sudo_for_tests_feature)
(download_cache,
reference_cloud_image_name) = requires_known_reference_image(self)
super(TestInstallWithSeed, self).setUp()
# We need to allow other users to read this dir
os.chmod(self.test_base_dir, 0755)
# We also need to sudo rm it as root created some files there
self.addCleanup(
svm.run_subprocess,
['sudo', 'rm', '-fr',
os.path.join(self.test_base_dir, 'home', '.virtinst')])
self.conf = svm.VmStack('selftest-seed')
self.vm = svm.Kvm(self.conf)
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir, 0755)
config_dir = os.path.join(self.test_base_dir, 'config')
self.conf.store._load_from_string('''
vm.name=selftest-seed
vm.update=False # Shorten install time
vm.cpus=2,
vm.release=raring
vm.config_dir=%s
vm.images_dir=%s
vm.download_cache=%s
vm.cloud_image_name=%s
vm.disk_size=8G
''' % (config_dir, images_dir, download_cache, reference_cloud_image_name))
def test_install_with_seed(self):
self.addCleanup(self.vm.undefine)
self.vm.install()
self.assertEqual('shut off', self.vm.state())
class TestInstallWithBacking(TestCaseWithHome):
def setUp(self):
tests.requires_feature(self, tests.use_sudo_for_tests_feature)
(download_cache_dir,
reference_cloud_image_name) = requires_known_reference_image(self)
super(TestInstallWithBacking, self).setUp()
# We need to allow other users to read this dir
os.chmod(self.test_base_dir, 0755)
# We also need to sudo rm it as root created some files there
self.addCleanup(
svm.run_subprocess,
['sudo', 'rm', '-fr',
os.path.join(self.test_base_dir, 'home', '.virtinst')])
self.conf = svm.VmStack('selftest-backing')
self.vm = svm.Kvm(self.conf)
# We'll share the images_dir between vms
images_dir = os.path.join(self.test_base_dir, 'images')
os.mkdir(images_dir, 0755)
config_dir = os.path.join(self.test_base_dir, 'config')
# Create a shared config
conf = svm.VmStack(None)
conf.store._load_from_string('''
vm.release=raring
vm.config_dir=%s
vm.images_dir=%s
vm.download_cache=%s
vm.disk_size=2G
vm.update=False # Shorten install time
[selftest-from-cloud]
vm.name=selftest-from-cloud
vm.cloud_image_name=%s
[selftest-backing]
vm.name=selftest-backing
vm.backing=selftest-from-cloud.qcow2
''' % (config_dir, images_dir, download_cache_dir, reference_cloud_image_name))
conf.store.save()
# Fake a previous install by just re-using the reference cloud image
temp_vm = svm.Kvm(svm.VmStack('selftest-from-cloud'))
temp_vm.create_disk_image()
def test_install_with_backing(self):
vm = svm.Kvm(svm.VmStack('selftest-backing'))
self.addCleanup(vm.undefine)
vm.install()
self.assertEqual('shut off', vm.state())
class TestSshKeyGen(TestCaseWithHome):
def setUp(self):
super(TestSshKeyGen, self).setUp()
self.conf = svm.VmStack(None)
self.vm = svm.VM(self.conf)
self.config_dir = os.path.join(self.test_base_dir, 'config')
def load_config(self, more):
content = '''\
vm.config_dir=%s
vm.name=foo
''' % (self.config_dir,)
self.conf.store._load_from_string(content + more)
def generate_key(self, ssh_type, upper_type=None):
if upper_type is None:
upper_type = ssh_type.upper()
self.load_config('vm.ssh_keys={vm.config_dir}/%s' % (ssh_type,))
self.vm.ssh_keygen()
private_path = 'config/%s' % (ssh_type,)
self.assertTrue(os.path.exists(private_path))
public_path = 'config/%s.pub' % (ssh_type,)
self.assertTrue(os.path.exists(public_path))
public = file(public_path).read()
private = file(private_path).read()
self.assertTrue(
private.startswith('-----BEGIN %s PRIVATE KEY-----\n'
% (upper_type,)))
self.assertTrue(
private.endswith('-----END %s PRIVATE KEY-----\n' % (upper_type,)))
return private, public
def test_dsa(self):
private, public = self.generate_key('dsa')
self.assertTrue(public.startswith('ssh-dss '))
self.assertTrue(public.endswith(' foo\n'))
def test_rsa(self):
private, public = self.generate_key('rsa')
self.assertTrue(public.startswith('ssh-rsa '))
self.assertTrue(public.endswith(' foo\n'))
def test_ecdsa(self):
tests.requires_feature(self, tests.ssh_feature)
tests.ssh_feature.requires_ecdsa(self)
private, public = self.generate_key('ecdsa', 'EC')
self.assertTrue(public.startswith('ecdsa-sha2-nistp256 '))
self.assertTrue(public.endswith(' foo\n'))
class TestOptionParsing(testtools.TestCase):
def setUp(self):
super(TestOptionParsing, self).setUp()
self.out = StringIO()
self.err = StringIO()
def parse_args(self, args):
return svm.arg_parser.parse_args(args, self.out, self.err)
def test_nothing(self):
self.assertRaises(SystemExit, self.parse_args, [])
def test_install(self):
ns = self.parse_args(['foo', '--install'])
self.assertEquals('foo', ns.name)
self.assertTrue(ns.install)
self.assertFalse(ns.download)
def test_download(self):
ns = self.parse_args(['foo', '--download'])
self.assertEquals('foo', ns.name)
self.assertFalse(ns.install)
self.assertTrue(ns.download)
class TestBuildCommands(TestCaseWithHome):
def setUp(self):
super(TestBuildCommands, self).setUp()
self.out = StringIO()
self.err = StringIO()
self.conf = svm.VmStack('foo')
self.conf.store._load_from_string('''\
[foo]
vm.name=foo
vm.class=lxc
''')
self.conf.store.save()
def build_commands(self, args):
return svm.build_commands(args, self.out, self.err)
def test_install(self):
cmds = self.build_commands(['--install', 'foo'])
self.assertEqual(1, len(cmds))
self.assertTrue(isinstance(cmds[0], svm.Install))
def test_download(self):
cmds = self.build_commands(['--download', 'foo'])
self.assertEqual(1, len(cmds))
self.assertTrue(isinstance(cmds[0], svm.Download))
def test_ssh_keygen(self):
cmds = self.build_commands(['--ssh-keygen', 'foo'])
self.assertEqual(1, len(cmds))
self.assertTrue(isinstance(cmds[0], svm.SshKeyGen))
def test_download_and_install(self):
cmds = self.build_commands(['--install', '--download', 'foo'])
self.assertEqual(2, len(cmds))
# Download comes first
self.assertTrue(isinstance(cmds[0], svm.Download))
self.assertTrue(isinstance(cmds[1], svm.Install))
class TestVmClass(testtools.TestCase):
def test_class_mandatory(self):
conf = svm.VmStack('I-dont-exist')
self.assertRaises(errors.ConfigOptionValueError, conf.get, 'vm.class')
def test_lxc(self):
conf = svm.VmStack('I-dont-exist')
conf.store._load_from_string('''vm.class=lxc''')
self.assertIs(svm.Lxc, conf.get('vm.class'))
def test_kvm(self):
conf = svm.VmStack('I-dont-exist')
conf.store._load_from_string('''vm.class=kvm''')
self.assertIs(svm.Kvm, conf.get('vm.class'))
def test_bogus(self):
conf = svm.VmStack('I-dont-exist')
conf.store._load_from_string('''vm.class=bogus''')
self.assertRaises(errors.ConfigOptionValueError, conf.get, 'vm.class')
# FIXME: This needs to be parametrized for KvmFromCloudImage and
# KvmFromBacking. Since we don't define vm.backing below, we're only testing
# KvmFromCloudImage for now. -- vila 2013-02-13
class TestInstall(TestCaseWithHome):
def setUp(self):
super(TestInstall, self).setUp()
self.conf = svm.VmStack('I-dont-exist')
self.conf.store._load_from_string('''
vm.name=I-dont-exist
vm.release=raring
vm.cpu_model=amd64
''')
self.states = []
def kvm_states(source=None):
return self.states
self.patch(svm, 'kvm_states', kvm_states)
self.vm = None
def install(self):
class FakeKvm(svm.Kvm):
def __init__(self, conf):
super(FakeKvm, self).__init__(conf)
self.undefine_called = False
self.install_called = False
# Make sure we avoid dangerous or costly calls
def poweroff(self):
pass
def undefine(self):
self.undefine_called = True
def install(self):
self.install_called = True
self.vm = FakeKvm(self.conf)
cmd = svm.Install(self.vm)
cmd.run()
def test_install_while_running(self):
self.conf.set('vm.name', 'foo')
self.states = {'foo': 'running'}
self.assertRaises(svm.SetupVmError, self.install)
self.assertFalse(self.vm.install_called)
self.assertFalse(self.vm.undefine_called)
def test_install_unknown(self):
self.states = {}
self.install()
self.assertTrue(self.vm.install_called)
self.assertFalse(self.vm.undefine_called)
def test_install_shutoff(self):
self.conf.set('vm.name', 'foo')
self.states = {'foo': 'shut off'}
self.install()
self.assertTrue(self.vm.install_called)
self.assertTrue(self.vm.undefine_called)
PK rIuƛ% % setup_vm/tests/__init__.pyimport errno
import os
import shutil
import subprocess
import tempfile
from bzrlib import osutils
def override_env_var(name, value):
"""Modify the environment, setting or removing the env_variable.
:param name: The environment variable to set.
:param value: The value to set the environment to. If None, then
the variable will be removed.
:return: The original value of the environment variable.
"""
orig = os.environ.get(name)
if value is None:
if orig is not None:
del os.environ[name]
else:
# FIXME: supporting unicode values requires a way to acquire the
# user encoding, punting for now -- vila 2013-01-30
os.environ[name] = value
return orig
def override_env(test, name, new):
"""Set an environment variable, and reset it after the test.
:param name: The environment variable name.
:param new: The value to set the variable to. If None, the variable is
deleted from the environment.
:returns: The actual variable value.
"""
value = override_env_var(name, new)
test.addCleanup(override_env_var, name, value)
return value
isolated_environ = {
'HOME': None,
}
def isolate_env(test, env=None):
"""Isolate test from the environment variables.
This is usually called in setUp for tests that needs to modify the
environment variables and restore them after the test is run.
:param test: A test instance
:param env: A dict containing variable definitions to be installed. Only
the variables present there are protected. They are initialized with
the provided values.
"""
if env is None:
env = isolated_environ
for name, value in env.items():
override_env(test, name, value)
def set_cwd_to_tmp(test):
"""Create a temp dir an cd into it for the test duration.
This is generally called during a test setup.
"""
test.test_base_dir = tempfile.mkdtemp(prefix='mytests-', suffix='.tmp')
test.addCleanup(shutil.rmtree, test.test_base_dir, True)
current_dir = os.getcwdu()
test.addCleanup(os.chdir, current_dir)
os.chdir(test.test_base_dir)
# FIXME: Borrowed from bzrlib, needs a better home -- vila 2013-08-05
class Feature(object):
"""An operating system Feature."""
def __init__(self):
self._available = None
def available(self):
"""Is the feature available?
:return: True if the feature is available.
"""
if self._available is None:
self._available = self._probe()
return self._available
def _probe(self):
"""Implement this method in concrete features.
:return: True if the feature is available.
"""
raise NotImplementedError
def __str__(self):
if getattr(self, 'feature_name', None):
return self.feature_name()
return self.__class__.__name__
class ExecutableFeature(Feature):
"""Feature testing whether an executable of a given name is on the PATH."""
def __init__(self, name):
super(ExecutableFeature, self).__init__()
self.name = name
self._path = None
@property
def path(self):
# This is a property, so accessing path ensures _probe was called
self.available()
return self._path
def _probe(self):
self._path = osutils.find_executable_on_path(self.name)
return self._path is not None
def feature_name(self):
return '%s executable' % self.name
qemu_img_feature = ExecutableFeature('qemu-img')
class _UseSudoForTestsFeature(Feature):
"""User has sudo access.
There is no direct way to test for sudo access other than trying to use
it. This is not something we can do in automated tests as it requires user
input.
Whatever trick is used to guess whether or not the user *can* sudo won't
tell us if she agrees to run the sudo tests. Instead, this should rely on
an opt-in mechanism so each user decides whether or not she wants to run
the tests.
"""
def _probe(self):
# I.e. if you want to run the tests that requires sudo issue:
# $ touch ~/.setup_vm.use_sudo_for_tests
# if you don't, issue:
# $ rm -f ~/.setup_vm.use_sudo_for_tests
path = os.path.expanduser('~/.setup_vm.use_sudo_for_tests')
return os.path.exists(path)
def feature_name(self):
return 'sudo access'
use_sudo_for_tests_feature = _UseSudoForTestsFeature()
class _SshFeature(ExecutableFeature):
def __init__(self):
super(_SshFeature, self).__init__('ssh')
self.version = None
def _probe(self):
exists = super(_SshFeature, self)._probe()
if exists:
try:
proc = subprocess.Popen(['ssh', '-V'],
stderr=subprocess.PIPE,
stdout=subprocess.PIPE)
out, err = proc.communicate()
except OSError as e:
if e.errno == errno.ENOENT:
# broken install
return False
else:
raise
self.version = err
return exists
def requires_ecdsa(self, test):
ecdsa_support = 'OpenSSH_5.9p1-5ubuntu.1.1'
if self.version < ecdsa_support:
test.skip('ecdsa requires ssh >= %s, you have %s'
% (ecdsa_support, self.version,))
ssh_feature = _SshFeature()
def requires_feature(test, feature):
if not feature.available():
test.skip('%s is not available' % feature.feature_name())
PK rI6Ys s setup_vm/tests/test_test.pyimport os
import testtools
from setup_vm import tests
def assertTestSuccess(test, inner):
"""The received test runs successfully."""
result = testtools.TestResult()
inner.run(result)
test.assertEqual(0, len(result.errors) + len(result.failures))
test.assertEqual(1, result.testsRun)
return result
class TestEnv(testtools.TestCase):
def test_env_preserved(self):
os.environ['NOBODY_USES_THIS'] = 'foo'
class Inner(testtools.TestCase):
def test_overridden(self):
tests.isolate_env(self, {'NOBODY_USES_THIS': 'bar'})
self.assertEqual('bar', os.environ['NOBODY_USES_THIS'])
assertTestSuccess(self, Inner('test_overridden'))
self.assertEqual('foo', os.environ['NOBODY_USES_THIS'])
def test_env_var_deleted(self):
os.environ['NOBODY_USES_THIS'] = 'foo'
class Inner(testtools.TestCase):
def test_deleted(self):
tests.isolate_env(self, {'NOBODY_USES_THIS': None})
self.assertIs('deleted',
os.environ.get('NOBODY_USES_THIS', 'deleted'))
assertTestSuccess(self, Inner('test_deleted'))
self.assertEqual('foo', os.environ['NOBODY_USES_THIS'])
class TestTmp(testtools.TestCase):
def test_cwd_in_tmp(self):
class Inner(testtools.TestCase):
def setUp(self):
super(Inner, self).setUp()
tests.set_cwd_to_tmp(self)
def test_cwd_in_tmp(self):
self.assertEqual(os.getcwdu(), self.test_base_dir)
assertTestSuccess(self, Inner('test_cwd_in_tmp'))
PK rIO6 setup_vm/bin/setup_vm.py#!/usr/bin/env python
"""
Setup a virtual machine from a config file.
Note: Most of the operations requires root access and this script uses ``sudo``
to get them.
"""
import argparse
from cStringIO import StringIO
import errno
import os
import subprocess
import sys
import bzrlib
from bzrlib import (
config,
registry,
transport,
urlutils,
)
import yaml
# Work around a bug in bzrlib.config forbidding some constructs in templates.
# Namely, spaces are invalid as an identifier and therefore should not match
# below.
config._option_ref_re = config.lazy_regex.lazy_compile('({[^ {},\n]+})')
class VmMatcher(config.NameMatcher):
def match(self, section):
if section.id is None:
# The no name section contains default values
return True
return super(VmMatcher, self).match(section)
def get_sections(self):
matching_sections = super(VmMatcher, self).get_sections()
return reversed(list(matching_sections))
class VmStore(config.LockableIniFileStore):
"""A config store for options specific to a directory."""
def __init__(self, directory, file_name, possible_transports=None):
t = transport.get_transport_from_path(
directory, possible_transports=possible_transports)
super(VmStore, self).__init__(t, file_name)
self.id = 'vm'
def system_config_dir():
return '/etc/libvirt'
class VmStack(config.Stack):
"""Per-directory options."""
def __init__(self, name):
"""Make a new stack for a given vm.
The following sections are queried:
* the ``name`` section in ./vms.conf,
* the no-name section in ./vms.conf
* the ``name`` section in ~/vms.conf,
* the no-name section in ~/vms.conf
* the ``name`` section in /etc/libvirt/vms.conf,
* the no-name section in /etc/libvirt/vms.conf
:param name: The name of a virtual machine.
"""
self.local_store = VmStore('.', 'vms.conf')
user_store = VmStore(os.environ['HOME'], 'vms.conf')
self.system_store = VmStore(system_config_dir(), 'vms.conf')
# FIXME: Only available in bzr-2.6b3 :-/ -- vila 2012-01-31
# dstore = self.get_shared_store()
super(VmStack, self).__init__(
[VmMatcher(self.local_store, name).get_sections,
VmMatcher(user_store, name).get_sections,
VmMatcher(self.system_store, name).get_sections,
],
user_store, mutable_section_id=name)
if bzrlib.version_info < (2, 6):
def get(self, name, expand=True):
"""Override base class to expand by default."""
return super(VmStack, self).get(name, expand)
def path_from_unicode(path_string):
if not isinstance(path_string, basestring):
raise TypeError
return os.path.expanduser(path_string)
class PathOption(config.Option):
def __init__(self, *args, **kwargs):
"""A path option definition.
This possibly expands the user home directory.
"""
super(PathOption, self).__init__(
*args, from_unicode=path_from_unicode, **kwargs)
if bzrlib.version_info < (2, 6):
class RegistryOption(config.Option):
"""Option for a choice from a registry."""
def __init__(self, name, registry, default_from_env=None,
help=None, invalid=None):
"""A registry based Option definition.
This overrides the base class so the conversion from a unicode
string can take quoting into account.
"""
super(RegistryOption, self).__init__(
name, default=lambda: unicode(registry.default_key),
default_from_env=default_from_env,
from_unicode=self.from_unicode, help=help,
invalid=invalid, unquote=False)
self.registry = registry
def from_unicode(self, unicode_str):
if not isinstance(unicode_str, basestring):
raise TypeError
try:
return self.registry.get(unicode_str)
except KeyError:
raise ValueError(
"Invalid value %s for %s."
"See help for a list of possible values."
% (unicode_str, self.name))
else:
RegistryOption = config.RegistryOption
# The VM classes are registered later (where they are defined)
vm_class_registry = registry.Registry()
def register(option):
config.option_registry.register(option)
register(config.Option('vm', default=None,
help='''The name space defining a virtual machine.
This option is a place holder to document the options that defines a virtual
machine and the options defining the infrastructure used to manage them all.
For qemu based vms, the definition of a vm is stored in an xml file under
'/etc/libvirt/qemu/{vm.name}.xml'. This is under the libvirt package control
and is out of scope for setup_vm.py.
There are 3 other significant files used for a given vm:
- a disk image mounted at '/' from '/dev/sda1':
'{vm.images_dir}/{vm.name}.qcow2'
- a iso image available from '/dev/sdb' labeled 'cidata':
{vm.images_dir}/{vm.name}.seed which contains the cloud-init data used to
configure/install/update the vm.
- a console: {vm.images_dir}/{vm.name}.console which can be 'tail -f'ed from
the host.
The data used to create the seed above are stored in a vm specific
configuration directory for easier debug and reference:
- {vm.config_dir}/user-data
- {vm.config_dir}/meta-data
- {vm.config_dir}/ecdsa
- {vm.config_dir}/ecdsa.pub
'''))
# The directory where we store vm files related to their configuration with
# cloud-init (user-data, meta-data, ssh keys).
register(config.Option('vm.vms_dir', default='~/.config/setup_vm',
help='''Where vm related config files are stored.
This includes user-data and meta-data for cloud-init and ssh server keys.
This directory must exist.
Each vm get a specific directory (automatically created) there based on its
name.
'''))
# The base directories where vms are stored for kvm
register(PathOption('vm.images_dir', default='/var/lib/libvirt/images',
help='''Where vm disk images are stored.''',))
register(config.Option('vm.qemu_etc_dir', default='/etc/libvirt/qemu',
help='''\
Where libvirt (qemu) stores the vms config files.'''))
# The base directories where vms are stored for lxc
register(PathOption('vm.lxcs_dir', default='/var/lib/lxc',
help='''Where lxc definitions are stored.'''))
# Isos and images download handling
register(config.Option('vm.iso_url',
default='http://cdimage.ubuntu.com/daily-live/current/',
help='''Where an iso can be downloaded from.'''))
register(config.Option('vm.iso_name',
default='{vm.release}-desktop-{vm.cpu_model}.iso',
help='''The name of the iso.'''))
register(config.Option('vm.cloud_image_url',
default='''\
http://cloud-images.ubuntu.com/{vm.release}/current/''',
help='''Where a cloud image can be downloaded from.'''))
register(config.Option('vm.cloud_image_name',
default='''\
{vm.release}-server-cloudimg-{vm.cpu_model}-disk1.img''',
help='''The name of the cloud image.'''))
register(PathOption('vm.download_cache', default='{vm.images_dir}',
help='''Where downloads end up.'''))
register(RegistryOption('vm.class', vm_class_registry,
invalid='error',
help='''The virtual machine technology to use.'''))
# The ubiquitous vm name
register(config.Option('vm.name', default=None, invalid='error',
help='''\
The vm name, used as a prefix for related files.'''))
# The second most important bit to define a vm: which ubuntu release ?
register(config.Option('vm.release', default=None, invalid='error',
help='''The ubuntu release name.'''))
# The third important piece to define a vm: where to store files like the
# console, the user-data and meta-data files, the ssh server keys, etc.
register(config.Option('vm.config_dir', default='{vm.vms_dir}/{vm.name}',
invalid='error',
help='''\
The directory where files specific to a vm are stored.
This includes the user-data and meta-data files used at install time (for
reference and easier debug) as well as the optional ssh server keys.
By default this is {vm.vms_dir}/{vm.name}. You can put it somewhere else by
redifining it as long as it ends up being unique for the vm.
{vm.vms_dir}/{vm.release}/{vm.name} may better suit your taste for example.
'''))
# The options defining the vm physical characteristics
register(config.Option('vm.ram_size', default='1024',
help='''The ram size in megabytes.'''))
register(config.Option('vm.disk_size', default='8G',
help='''The disk image size in bytes.
Optional suffixes "k" or "K" (kilobyte, 1024) "M" (megabyte, 1024k) "G"
(gigabyte, 1024M) and T (terabyte, 1024G) are supported.
'''))
register(config.Option('vm.cpus', default='1', help='''The number of cpus.'''))
register(config.Option('vm.cpu_model', default=None, invalid='error',
help='''The number of cpus.'''))
register(config.Option('vm.network', default='network=default',
invalid='error', help='''\
The --network parameter for virt-install.
This can be specialized for each machine but the default should work in most
setups. Watch for your DHCP server exhausting its address space if you create a
lot of vms with random MAC addresses.
'''))
register(config.Option('vm.meta_data', default='''\
instance-id: {vm.name}
local-hostname: {vm.name}
''',
invalid='error',
help='''\
The meta data for cloud-init to put in the seed.'''))
# Some bits that may added to user-data but are optional
register(config.ListOption('vm.packages', default=None,
help='''\
A list of package names to be installed.'''))
register(config.Option('vm.apt_proxy', default=None, invalid='error',
help='''\
A local proxy for apt to avoid repeated .deb downloads.
Example:
vm.apt_proxy = http://192.168.0.42:8000
'''))
register(config.ListOption('vm.apt_sources', default=None,
help='''\
A list of apt sources entries to be added to the default ones.
Cloud-init already setup /etc/apt/sources.list with appropriate entries. Only
additional entries need to be specified here.
'''))
register(config.ListOption('vm.ssh_authorized_keys', default=None,
help='''\
A list of paths to public ssh keys to be authorized for the default user.'''))
register(config.ListOption('vm.ssh_keys', default=None,
help='''A list of paths to server ssh keys.
Both public and private keys can be provided. Accepted ssh key types are rsa,
dsa and ecdsa. The file names should match .*[.pub].
'''))
register(config.Option('vm.update', default=False,
from_unicode=config.bool_from_store,
help='''Whether or not the vm should be updated.
Both apt-get update and apt-get upgrade are called if this option is set.
'''))
register(config.Option('vm.password', default='ubuntu', invalid='error',
help='''The ubuntu user password.'''))
register(config.Option('vm.launchpad_id',
help='''\
The launchpad login used for launchpad ssh access from the guest.'''))
# The scripts that are executed before powering off
register(PathOption('vm.root_script', default=None,
help='''\
The path to a script executed as root before powering off.
This script is executed before {vm.ubuntu_script}.
'''))
register(PathOption('vm.ubuntu_script', default=None,
help='''\
The path to a script executed as ubuntu before powering off.
This script is excuted after {vm.root_script}.
'''))
register(config.ListOption('vm.uploaded_scripts', default=None,
help='''\
A list of scripts to be uploaded to the guest.
Scripts can use config options from their vm, they will be expanded before
upload. All scripts are uploaded into {vm.uploaded_scripts.guest_dir} under
their base name.
'''))
register(config.Option('vm.uploaded_scripts.guest_dir',
default='~ubuntu/bin',
help='''\
Where {vm.uploaded_scripts} are uploaded on the guest.'''))
class SetupVmError(Exception):
msg = 'setup_vm Generic Error: %r'
def __init__(self, msg=None, **kwds):
if msg is not None:
self.msg = msg
for key, value in kwds.items():
setattr(self, key, value)
def __str__(self):
return self.msg.format((), **self.__dict__)
__repr__ = __str__
class CommandError(SetupVmError):
msg = '''
command: {joined_cmd}
retcode: {retcode}
output: {out}
error: {err}
'''
def __init__(self, cmd, retcode, out, err):
super(CommandError, self).__init__(joined_cmd=' '.join(cmd),
retcode=retcode, err=err, out=out)
class ConfigValueError(SetupVmError):
msg = 'Bad value "{value}" for option "{name}".'
def __init__(self, name, value):
super(ConfigValueError, self).__init__(name=name, value=value)
class ConfigPathNotFound(SetupVmError):
msg = 'No such file: {path} from {name}'
def __init__(self, path, name):
super(ConfigPathNotFound, self).__init__(path=path, name=name)
def run_subprocess(args):
proc = subprocess.Popen(args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.PIPE)
out, err = proc.communicate()
if proc.returncode:
raise CommandError(args, proc.returncode, out, err)
return proc.returncode, out, err
def pipe_subprocess(args):
proc = subprocess.Popen(args,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
return proc
def ssh_infos_from_path(key_path):
"""Analyze path to find ssh key type and kind.
The basename should begin with ssh type used to create the key. and end
with '.pub' for a public key.
If the type is neither of rds, dsa or ecdsa, None if returned.
:param key_path: A path to an ssh key.
:return: (type, kind) 'type' is the ssh key type or None if neither of rds,
dsa or ecdsa. 'kind' is 'public' if the path ends with '.pub',
'private' otherwise.
"""
base = os.path.basename(key_path)
for p in ('rsa', 'dsa', 'ecdsa'):
if base.startswith(p):
ssh_type = p
break
else:
ssh_type = None
if base.endswith('.pub'):
kind = 'public'
else:
kind = 'private'
return ssh_type, kind
class ConsoleEOFError(SetupVmError):
msg = 'Encountered EOF on console, something went wrong'
class CloudInitError(SetupVmError):
msg = 'cloud-init reported: {line} check your config'
def __init__(self, line):
super(CloudInitError, self).__init__(line=line)
class ConsoleMonitor(object):
"""Monitor a console to identify known events."""
def __init__(self, stream):
super(ConsoleMonitor, self).__init__()
self.stream = stream
def parse(self):
while True:
line = self.stream.readline()
yield line
if not line:
raise ConsoleEOFError()
elif line.startswith(' * Will now halt'):
# That's our final_message, we're done
return
elif ('Failed loading yaml blob' in line or
'Unhandled non-multipart userdata starting' in line or
'failed to render string to stdout:' in line or
'Failed loading of cloud config' in line):
raise CloudInitError(line)
class FileMonitor(ConsoleMonitor):
def __init__(self, path):
cmd = ['tail', '-f', path]
proc = pipe_subprocess(cmd)
super(FileMonitor, self).__init__(proc.stdout)
self.path = path
self.cmd = cmd
self.proc = proc
self.lines = []
def parse(self):
try:
for line in super(FileMonitor, self).parse():
self.lines.append(line)
yield line
finally:
self.proc.terminate()
class CIUserData(object):
"""Maps configuration data into cloud-init user-data.
This is a container for the data that will ultimately be encoded into a
cloud-config-archive user-data file.
"""
def __init__(self, conf):
super(CIUserData, self).__init__()
self.conf = conf
# The objects we will populate before creating a yaml encoding as a
# cloud-config-archive file
self.cloud_config = {}
self.root_hook = None
self.ubuntu_hook = None
self.launchpad_hook = None
self.uploaded_scripts_hook = None
def set(self, ud_name, conf_name=None, value=None):
"""Set a user-data option from it's corresponding configuration one.
:param ud_name: user-data key.
:param conf_name: configuration key, If set to None, `value` should be
provided.
:param value: value to use if `conf_name` is None.
"""
if value is None and conf_name is not None:
value = self.conf.get(conf_name)
if value is not None:
self.cloud_config[ud_name] = value
def _file_content(self, path, option_name):
full_path = os.path.expanduser(path)
try:
with open(full_path) as f:
content = f.read()
except IOError, e:
if e.args[0] == errno.ENOENT:
raise ConfigPathNotFound(path, option_name)
else:
raise
return content
def set_list_of_paths(self, ud_name, conf_name):
"""Set a user-data option from its corresponding configuration one.
The configuration option is a list of paths and the user-data option
will be a list of each file content.
:param ud_name: user-data key.
:param conf_name: configuration key.
"""
paths = self.conf.get(conf_name)
if paths:
contents = []
for p in paths:
contents.append(self._file_content(p, conf_name))
self.set(ud_name, None, contents)
def _key_from_path(self, path, option_name):
"""Infer user-data key from file name."""
ssh_type, kind = ssh_infos_from_path(path)
if ssh_type is None:
raise ConfigValueError(option_name, path)
return '%s_%s' % (ssh_type, kind)
def set_ssh_keys(self):
"""Set the server ssh keys from a list of paths.
Provided paths should respect some coding:
- the base name should start with the ssh type of their key (rsa, dsa,
ecdsa),
- base names ending with '.pub' are for public keys, the others are for
private keys.
"""
key_paths = self.conf.get('vm.ssh_keys')
if key_paths:
ssh_keys = {}
for p in key_paths:
key = self._key_from_path(p, 'vm.ssh_keys')
ssh_keys[key] = self._file_content(p, 'vm.ssh_keys')
self.set('ssh_keys', None, ssh_keys)
def set_apt_sources(self):
sources = self.conf.get('vm.apt_sources')
if sources:
apt_sources = []
for src in sources:
# '|' should not appear in urls nor keys so it should be safe
# to use it as a separator.
parts = src.split('|')
if len(parts) == 1:
apt_sources.append({'source': parts[0]})
else:
# For PPAs, an additional GPG key should be imported in the
# guest.
apt_sources.append({'source': parts[0], 'keyid': parts[1]})
self.cloud_config['apt_sources'] = apt_sources
def append_cmd(self, cmd):
cmds = self.cloud_config.get('runcmd', [])
cmds.append(cmd)
self.cloud_config['runcmd'] = cmds
def _hook_script_path(self, user):
return '~%s/setup_vm_post_install' % (user,)
def _hook_content(self, option_name, user, hook_path, mode='0700'):
# FIXME: Add more tests towards properly creating a tree on the guest
# from a tree on the host. There seems to be three kind of items worth
# caring about here: file content (output path, owner, chmod), file
# (input and output paths, owner, chmod) and directory (path, owner,
# chmod). There are also some subtle traps involved about when files
# are created across various vm generations (one vm creates a dir, a mv
# on top of that one doesn't, but still creates a file in this dir,
# without realizing it can fail in a fresh vm). -- vila 2013-03-10
host_path = self.conf.get(option_name)
if host_path is None:
return None
fcontent = self._file_content(host_path, option_name)
# Expand options in the provided content so we report better errors
expanded_content = self.conf.expand_options(fcontent)
# The following will generate an additional newline at the end of the
# script. I can't think of a case where it matters and it makes this
# code more robust (and/or simpler) if the script/file *doesn't* end up
# with a proper newline.
# FIXME: This may be worth fixing if we provide a more generic way to
# create a remote tree. -- vila 2013-03-10
hook_content = '''#!/bin/sh
cat >{__guest_path} <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
{__fcontent}
EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown {__user}:{__user} {__guest_path}
chmod {__mode} {__guest_path}
'''
return hook_content.format(__user=user, __fcontent=expanded_content,
__mode=mode,
__guest_path=hook_path)
def set_boot_hook(self):
# FIXME: Needs a test ensuring we execute as root -- vila 2013-03-07
hook_path = self._hook_script_path('root')
content = self._hook_content('vm.root_script', 'root', hook_path)
if content is not None:
self.root_hook = content
self.append_cmd(hook_path)
def set_ubuntu_hook(self):
# FIXME: Needs a test ensuring we execute as ubuntu -- vila 2013-03-07
hook_path = self._hook_script_path('ubuntu')
content = self._hook_content('vm.ubuntu_script', 'ubuntu', hook_path)
if content is not None:
self.ubuntu_hook = content
self.append_cmd(['su', '-l', '-c', hook_path, 'ubuntu'])
def set_launchpad_access(self):
# FIXME: Needs a test that we can really access launchpad properly via
# ssh. Can only be done as a real launchpad user and as such requires
# cooperation :) I.e. Some configuration option set by the user will
# trigger the test -- vila 2013-03-14
lp_id = self.conf.get('vm.launchpad_id')
if lp_id is None:
return
# Use the specified ssh key found in ~/.ssh as the private key. The
# user is supposed to have uploaded the public one.
local_path = os.path.join('~', '.ssh', '%s@setup_vm' % (lp_id,))
# Force id_rsa or we'll need a .ssh/config to point to user@setup_vm
# for .lauchpad.net.
hook_path = '/home/ubuntu/.ssh/id_rsa'
dir_path = os.path.dirname(hook_path)
try:
fcontent = self._file_content(local_path, 'vm.launchpad_id')
except ConfigPathNotFound, e:
e.msg = ('You need to create the {p} keypair and upload {p}.pub to'
' launchpad.\n'
'See vm.launchpad_id in README.'.format(p=local_path))
raise e
# FIXME: ~Duplicated from _hook_content. -- vila 2013-03-10
# FIXME: If this hook is executed before the ubuntu user is created we
# need to chown/chmod ~ubuntu which is bad. This happens when a
# -pristine vm is created and lead to GUI login failing because it
# can't create any dir/file there. The fix is to only create a script
# that will be executed via runcmd so it will run later and avoid the
# issue. -- vila 2013-03-21
# FIXME: Moreover, -pristine vms don't have bzr installed so this
# cannot succeed there -- vila 2013-08-07
hook_content = '''#!/bin/sh
mkdir -p {dir_path}
chown {user}:{user} ~ubuntu
chmod {dir_mode} ~ubuntu
chown {user}:{user} {dir_path}
chmod {dir_mode} {dir_path}
cat >{guest_path} <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
{fcontent}
EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown {user}:{user} {guest_path}
chmod {file_mode} {guest_path}
'''
self.launchpad_hook = self.conf.expand_options(
hook_content,
env=dict(user='ubuntu', fcontent=fcontent,
file_mode='0400', guest_path=hook_path,
dir_mode='0700', dir_path=dir_path))
self.append_cmd(['sudo', '-u', 'ubuntu',
'bzr', 'launchpad-login', lp_id])
def set_uploaded_scripts(self):
script_paths = self.conf.get('vm.uploaded_scripts')
if not script_paths:
return
hook_path = '~ubuntu/setup_vm_uploads'
bindir = self.conf.get('vm.uploaded_scripts.guest_dir')
out = StringIO()
out.write('''#!/bin/sh
cat >{hook_path} <<'EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN'
mkdir -p {bindir}
cd {bindir}
'''.format(**locals()))
for path in script_paths:
fcontent = self._file_content(path, 'vm.uploaded_scripts')
expanded = self.conf.expand_options(fcontent)
base = os.path.basename(path)
# FIXME: ~Duplicated from _hook_content. -- vila 2012-03-15
out.write('''cat >{base} <<'EOF{base}'
{expanded}
EOF{base}
chmod 0755 {base}
'''.format(**locals()))
out.write('''EOSETUPVMUNIQUECONTENTDONTBREAKFORFUN
chown {user}:{user} {hook_path}
chmod 0700 {hook_path}
'''.format(user='ubuntu', **locals()))
self.uploaded_scripts_hook = out.getvalue()
self.append_cmd(['su', '-l', '-c', hook_path, 'ubuntu'])
def set_poweroff(self):
# We want to shutdown properly after installing. This is safe to set
# here as subsequent boots will ignore this setting, letting us use the
# vm ;)
if self.conf.get('vm.release') in ('precise', 'quantal'):
# Curse cloud-init lack of compatibility
self.append_cmd('halt')
else:
self.set('power_state', None, {'mode': 'poweroff'})
def populate(self):
# Common and non-configurable options
if self.conf.get('vm.release') == 'precise':
# Curse cloud-init lack of compatibility
msg = 'setup_vm finished installing in $UPTIME seconds.'
else:
msg = 'setup_vm finished installing in ${uptime} seconds.'
self.set('final_message', None, msg)
self.set('manage_etc_hosts', None, True)
self.set('chpasswd', None, dict(expire=False))
# Configurable options
self.set('password', 'vm.password')
self.set_list_of_paths('ssh_authorized_keys', 'vm.ssh_authorized_keys')
self.set_ssh_keys()
self.set('apt_proxy', 'vm.apt_proxy')
# Both user-data keys are set from the same config key, we don't
# provide a finer access.
self.set('apt_update', 'vm.update')
self.set('apt_upgrade', 'vm.update')
self.set_apt_sources()
self.set('packages', 'vm.packages')
self.set_launchpad_access()
# uploaded scripts
self.set_uploaded_scripts()
# The commands executed before powering off
self.set_boot_hook()
self.set_ubuntu_hook()
# This must be called last so previous commands (for precise and
# quantal) can be executed before powering off
self.set_poweroff()
def add_boot_hook(self, parts, hook):
if hook is not None:
parts.append({'content': '#cloud-boothook\n' + hook})
def dump(self):
parts = [{'content': '#cloud-config\n' +
yaml.safe_dump(self.cloud_config)}]
self.add_boot_hook(parts, self.root_hook)
self.add_boot_hook(parts, self.ubuntu_hook)
self.add_boot_hook(parts, self.launchpad_hook)
self.add_boot_hook(parts, self.uploaded_scripts_hook)
# Wrap the lot into a cloud config archive
return '#cloud-config-archive\n' + yaml.safe_dump(parts)
class VM(object):
"""A virtual machine relying on cloud-init to customize installation."""
def __init__(self, conf):
self.conf = conf
self._config_dir = None
# Seed files
self._meta_data_path = None
self._user_data_path = None
def _download_in_cache(self, source_url, name, force=False):
"""Download ``name`` from ``source_url`` in ``vm.download_cache``.
:param source_url: The url where the file to download is located
:param name: The name of the file to download (also used as the name
for the downloaded file).
:param force: Remove the file from the cache if present.
:return: False if the file is in the download cache, True if a download
occurred.
"""
source = urlutils.join(source_url, name)
download_dir = self.conf.get('vm.download_cache')
if not os.path.exists(download_dir):
raise ConfigValueError('vm.download_cache', download_dir)
target = os.path.join(download_dir, name)
# FIXME: By default the download dir may be under root control, but if
# a user chose to use a different one under his own control, it would
# be nice to not require sudo usage. -- vila 2013-02-06
if force:
run_subprocess(['sudo', 'rm', '-f', target])
if not os.path.exists(target):
# FIXME: We do ask for a progress bar but it's not displayed
# (run_subprocess capture both stdout and stderr) ! At least while
# used interactively, it should. -- vila 2013-02-06
run_subprocess(['sudo', 'wget', '--progress=dot:mega', '-O',
target, source])
return True
else:
return False
def ensure_dir(self, path):
try:
os.mkdir(path)
except OSError, e:
# FIXME: Try to create the parent dir ?
if e.errno == errno.EEXIST:
pass
else:
raise
def ensure_config_dir(self):
if self._config_dir is None:
# FIXME: expanduser is not tested
self._config_dir = os.path.expanduser(
self.conf.get('vm.config_dir'))
self.ensure_dir(self._config_dir)
def _ssh_keygen(self, key_path):
ssh_type, kind = ssh_infos_from_path(key_path)
path = os.path.expanduser(key_path) # Just in case
if kind == 'private': # public will be generated at the same time
run_subprocess(
['ssh-keygen', '-f', path, '-N', '', '-t', ssh_type,
'-C', self.conf.get('vm.name')])
def ssh_keygen(self):
self.ensure_config_dir()
keys = self.conf.get('vm.ssh_keys')
for key in keys:
self._ssh_keygen(key)
def create_meta_data(self):
self.ensure_config_dir()
self._meta_data_path = os.path.join(self._config_dir, 'meta-data')
with open(self._meta_data_path, 'w') as f:
f.write(self.conf.get('vm.meta_data'))
def create_user_data(self):
ci_user_data = CIUserData(self.conf)
ci_user_data.populate()
self.ensure_config_dir()
self._user_data_path = os.path.join(self._config_dir, 'user-data')
with open(self._user_data_path, 'w') as f:
f.write(ci_user_data.dump())
def download(self, force=False):
raise NotImplementedError(self.download)
def parse_console_during_install(self, cmd):
"""Parse the console output until the end of the install.
We added a specific part for cloud-init to ensure we properly detect
the end of the run.
:param cmd: The install command (used for error display).
"""
console = FileMonitor(self._console_path)
try:
for line in console.parse():
# FIXME: We need some way to activate this dynamically (conf
# var defaulting to env var OR cmdline parameter ?
# -- vila 2013-02-11
# print "read: [%s]" % (line,) # so useful for debug...
pass
except (ConsoleEOFError, CloudInitError):
# FIXME: No test covers this path -- vila 2013-02-15
err_lines = ['Suspicious line from cloud-init.\n',
'\t' + console.lines[-1],
'Check the configuration:\n']
with open(self._meta_data_path) as f:
err_lines.append('meta-data content:\n')
err_lines.extend(f.readlines())
with open(self._user_data_path) as f:
err_lines.append('user-data content:\n')
err_lines.extend(f.readlines())
raise CommandError(cmd, console.proc.returncode,
'\n'.join(console.lines),
''.join(err_lines))
def kvm_states(source=None):
"""A dict of states for kvms indexed by name.
:param source: A list of lines as produced by virsh list --all without
decorations (header/footer).
"""
if source is None:
retcode, out, err = run_subprocess(['virsh', 'list', '--all'])
# Get rid of header/footer
source = out.splitlines()[2:-1]
states = {}
for line in source:
caret_or_id, name, state = line.split(None, 2)
states[name] = state
return states
class Kvm(VM):
def __init__(self, conf):
super(Kvm, self).__init__(conf)
# Disk paths
self._disk_image_path = None
self._seed_path = None
self._console_path = None
def state(self):
states = kvm_states()
try:
state = states[self.conf.get('vm.name')]
except KeyError:
state = None
return state
def download_iso(self, force=False):
"""Download the iso to install the vm.
:return: False if the iso is in the download cache, True if a download
occurred.
"""
return self._download_in_cache(self.conf.get('vm.iso_url'),
self.conf.get('vm.iso_name'),
force=force)
def download_cloud_image(self, force=False):
"""Download the cloud image to install the vm.
:return: False if the image is in the download cache, True if a
download occurred.
"""
return self._download_in_cache(self.conf.get('vm.cloud_image_url'),
self.conf.get('vm.cloud_image_name'),
force=force)
def download(self, force=False):
return self.download_cloud_image(force)
def create_seed_image(self):
if self._meta_data_path is None:
self.create_meta_data()
if self._user_data_path is None:
self.create_user_data()
images_dir = self.conf.get('vm.images_dir')
seed_path = os.path.join(
images_dir, self.conf.expand_options('{vm.name}.seed'))
run_subprocess(
# We create the seed in the ``vm.images_dir`` directory, so
# ``sudo`` is required
['sudo',
'genisoimage', '-output', seed_path,
# cloud-init relies on the volid to discover its data
'-volid', 'cidata',
'-joliet', '-rock', '-input-charset', 'default',
'-graft-points',
'user-data=%s' % (self._user_data_path,),
'meta-data=%s' % (self._meta_data_path,),
])
self._seed_path = seed_path
def create_disk_image(self):
if self.conf.get('vm.backing') is None:
self.create_disk_image_from_cloud_image()
else:
self.create_disk_image_from_backing()
def create_disk_image_from_cloud_image(self):
"""Create a disk image from a cloud one."""
cloud_image_path = os.path.join(
self.conf.get('vm.download_cache'),
self.conf.get('vm.cloud_image_name'))
disk_image_path = os.path.join(
self.conf.get('vm.images_dir'),
self.conf.expand_options('{vm.name}.qcow2'))
run_subprocess(
['sudo', 'qemu-img', 'convert',
'-O', 'qcow2', cloud_image_path, disk_image_path])
run_subprocess(
['sudo', 'qemu-img', 'resize',
disk_image_path, self.conf.get('vm.disk_size')])
self._disk_image_path = disk_image_path
def create_disk_image_from_backing(self):
"""Create a disk image backed by an existing one."""
backing_image_path = os.path.join(
self.conf.get('vm.images_dir'),
self.conf.expand_options('{vm.backing}'))
disk_image_path = os.path.join(
self.conf.get('vm.images_dir'),
self.conf.expand_options('{vm.name}.qcow2'))
run_subprocess(
['sudo', 'qemu-img', 'create', '-f', 'qcow2',
'-b', backing_image_path, disk_image_path])
run_subprocess(
['sudo', 'qemu-img', 'resize',
disk_image_path, self.conf.get('vm.disk_size')])
self._disk_image_path = disk_image_path
def parse_console_during_install(self, cmd):
"""See Vm.parse_console_during_install."""
# The console is created by virt-install which requires sudo but
# creates the file 0600 for libvirt-qemu. We give read access to all
# otherwise 'tail -f' requires sudo and can't be killed anymore.
run_subprocess(['sudo', 'chmod', '0644', self._console_path])
# While `virt-install` is running, let's connect to the console
super(Kvm, self).parse_console_during_install(cmd)
def install(self):
# Create a kvm, relying on cloud-init to customize the base image.
#
# There are two processes involvded here:
# - virt-install creates the vm and boots it.
# - progress is monitored via the console to detect cloud-final.
#
# Once cloud-init has finished, the vm can be powered off.
# FIXME: If the install doesn't finish after $time, emit a warning and
# terminate self.install_proc.
# FIXME: If we can't connect to the console, emit a warning and
# terminate console and self.install_proc.
# FIXME: If we don't receive anything on the console after $time2, emit
# a warning and terminate console and self.install_proc.
# -- vila 2013-02-07
if self._seed_path is None:
self.create_seed_image()
if self._disk_image_path is None:
self.create_disk_image()
# FIXME: Install time is probably a good time to delete the
# console. While it makes sense to accumulate for all runs for a given
# install, keeping them without any limit nor roration is likely to
# cause issues at some point... -- vila 2013-02-20
self._console_path = os.path.join(
self.conf.get('vm.images_dir'),
'%s.console' % (self.conf.get('vm.name'),))
virt_install = [
'sudo', 'virt-install',
# To ensure we're not bitten again by http://pad.lv/1157272 where
# virt-install wrongly detect virtualbox. -- vila 2013-03-20
'--connect', 'qemu:///system',
# Without --noautoconsole, virt-install will relay the console,
# that's not appropriate for our needs so we'll connect later
# ourselves
'--noautoconsole',
# We define the console as a file so we can monitor the install
# via 'tail -f'
'--serial', 'file,path=%s' % (self._console_path,),
'--network', self.conf.get('vm.network'),
# Anticipate that we'll need a graphic card defined
'--graphics', 'spice',
'--name', self.conf.get('vm.name'),
'--ram', self.conf.get('vm.ram_size'),
'--vcpus', self.conf.get('vm.cpus'),
'--disk', 'path=%s,format=qcow2' % (self._disk_image_path,),
'--disk', 'path=%s' % (self._seed_path,),
# We just boot, cloud-init will handle the installs we need
'--boot', 'hd', '--hvm',
]
run_subprocess(virt_install)
self.parse_console_during_install(virt_install)
# We've seen the console signaling halt, but the vm will need a bit
# more time to get there so we help it a bit.
if self.conf.get('vm.release') in ('precise', 'quantal'):
# cloud-init doesn't implement power_state until raring and need a
# gentle nudge.
self.poweroff()
vm_name = self.conf.get('vm.name')
while True:
state = self.state()
# We expect the vm's state to be 'in shutdown' but in some rare
# occasions we may catch 'running' before getting 'in shutdown'.
if state in ('in shutdown', 'running'):
# Ok, querying the state takes time, this regulates the polling
# enough that we don't need to sleep.
continue
elif state == 'shut off':
# Good, we're done
break
# FIXME: No idea on how to test the following. Manually tested by
# altering the expected state above and running 'selftest.py -v'
# which errors out for test_install_with_seed and
# test_install_backing. Also reproduced when 'running' wasn't
# expected before 'in shutdown' -- vila 2013-02-19
# Unexpected state reached, bad.
raise SetupVmError('Something went wrong during {name} install\n'
'The vm ended in state: {state}\n'
'Check the console at {path}',
name=vm_name, state=state,
path=self._console_path)
def poweroff(self):
return run_subprocess(
['sudo', 'virsh', 'destroy', self.conf.get('vm.name')])
def undefine(self):
return run_subprocess(
['sudo', 'virsh', 'undefine', self.conf.get('vm.name'),
'--remove-all-storage'])
vm_class_registry.register('kvm', Kvm, 'Kernel-based virtual machine')
def lxc_info(vm_name, source=None):
"""Parse state info from the lxc-info output.
:param vm_name: The vm we want to query about.
:param source: A list of lines as produced by virsh list --all without
decorations (header/footer).
"""
if source is None:
retcode, out, err = run_subprocess(['sudo', 'lxc-info', '-n', vm_name])
source = out.splitlines()
state_line, pid_line = source
_, state = state_line.split(None, 1)
_, pid = pid_line.split(None, 1)
return dict(state=state, pid=pid)
class Lxc(VM):
def __init__(self, conf):
super(Lxc, self).__init__(conf)
self._guest_seed_path = None
self._fstab_path = None
def state(self):
info = lxc_info(self.conf.get('vm.name'))
return info['state']
def download(self, force=False):
# FIXME: lxc-create provides its own cache. download(True) should just
# ensure we clear that cache from the previous download. Should we add
# a warning ? Specialize the cache for Kvm only ?-- vila 2013-08-07
return True
def create_seed_files(self):
if self._meta_data_path is None:
self.create_meta_data()
if self._user_data_path is None:
self.create_user_data()
self._fstab_path = os.path.join(self._config_dir, 'fstab')
self._guest_seed_path = os.path.join(
self.conf.get('vm.lxcs_dir'),
self.conf.get('vm.name'),
'rootfs/var/lib/cloud/seed/nocloud-net')
with open(self._fstab_path, 'w') as f:
# Add a entry so cloud-init find the seed files
f.write('%s %s none bind 0 0\n' % (self._config_dir,
self._guest_seed_path))
def install(self):
'''Create an lxc, relying on cloud-init to customize the base image.
There are two processes involvded here:
- lxc-create creates the vm.
- progress is monitored via the console to detect cloud-final.
Once cloud-init has finished, the vm can be powered off.
'''
# FIXME: If the install doesn't finish after $time, emit a warning and
# terminate self.install_proc.
# FIXME: If we can't connect to the console, emit a warning and
# terminate console and self.install_proc.
# FIXME: If we don't receive anything on the console after $time2, emit
# a warning and terminate console and self.install_proc.
# -- vila 2013-02-07
if self._fstab_path is None:
self.create_seed_files()
# FIXME: Install time is probably a good time to delete the
# console. While it makes sense to accumulate for all runs for a given
# install, keeping them without any limit nor roration is likely to
# cause issues at some point... -- vila 2013-02-20
self._console_path = os.path.join(
# FIXME: We use _config_dir instead of 'vm.images_dir' as kvm does
# because the later is owned by root so we can't create a file
# there. It would be nice to check if the same trick can be used
# for kvm to simplify. -- vila 2013-08-07
self._config_dir,
'%s.console' % (self.conf.get('vm.name'),))
# Create/empty the file so we get access to it (otherwise it will be
# owned by root).
open(self._console_path, 'w').close()
# FIXME: Some feedback would be nice during lxc creation, not sure
# about which errors to expect there either -- vila 2013-08-07
run_subprocess(
['sudo', 'lxc-create',
'-n', self.conf.get('vm.name'),
'-t', 'ubuntu-cloud',
'--',
'-r', self.conf.get('vm.release'),
'-a', self.conf.get('vm.cpu_model'),
'-C', # From cloud image, implying download/cache
])
# Now we add the cloud-init data seed and do lxc-start to trigger all
# our customizations monitoring the lxc-start output from the host.
mkdir_seed_path = 'mkdir -p %s' % (self._guest_seed_path,)
lxc_start = ['sudo', 'lxc-start',
'-n', self.conf.get('vm.name'),
'--define', 'lxc.hook.pre-start=%s' % (mkdir_seed_path,),
'--define', 'lxc.mount=%s' % (self._fstab_path,),
'--console-log', self._console_path,
# Daemonize or: 1) it fails with a spurious return code,
# 2) We can't monitor the logfile
'-d',
]
run_subprocess(lxc_start)
self.parse_console_during_install(lxc_start)
def poweroff(self):
return run_subprocess(
['sudo', 'lxc-stop', '-n', self.conf.get('vm.name')])
def undefine(self):
try:
return run_subprocess(
['sudo', 'lxc-destroy', '-n', self.conf.get('vm.name')])
except CommandError as e:
# FIXME: No test -- vila 2013-08-08
if e.err.endswith('does not exist\n'):
# Fine. lxc-info makes no distinction between a stopped vm and
# a non-existing one.
pass
else:
raise
vm_class_registry.register('lxc', Lxc, 'Linux container virtual machine')
class ArgParser(argparse.ArgumentParser):
"""A parser for the setup_vm script."""
def __init__(self):
description = 'Set up virtual machines from a configuration file.'
super(ArgParser, self).__init__(
prog='setup_vm.py', description=description)
self.add_argument(
'name', help='Virtual machine section in the configuration file.')
self.add_argument('--download', '-d', action="store_true",
help='Force download of the required image.')
self.add_argument('--ssh-keygen', '-k', action="store_true",
help='Generate the ssh server keys (if any).')
self.add_argument('--install', '-i', action="store_true",
help='Install the virtual machine.')
def parse_args(self, args=None, out=None, err=None):
"""Parse arguments, overridding stdout/stderr if provided.
Overridding stdout/stderr is provided for tests.
:params args: Defaults to sys.argv[1:].
:param out: Defaults to sys.stdout.
:param err: Defaults to sys.stderr.
"""
out_orig = sys.stdout
err_orig = sys.stderr
try:
if out is not None:
sys.stdout = out
if err is not None:
sys.stderr = err
return super(ArgParser, self).parse_args(args)
finally:
sys.stdout = out_orig
sys.stderr = err_orig
arg_parser = ArgParser()
class Command(object):
def __init__(self, vm):
self.vm = vm
class Download(Command):
def run(self):
self.vm.download(force=True)
class SshKeyGen(Command):
def run(self):
self.vm.ssh_keygen()
class Install(Command):
def run(self):
vm_name = self.vm.conf.get('vm.name')
state = self.vm.state()
if state in('shut off', 'STOPPED'):
self.vm.undefine()
elif state in ('running', 'RUNNING'):
raise SetupVmError('{name} is running', name=vm_name)
self.vm.install()
def build_commands(args=None, out=None, err=None):
cmds = []
if args is None:
args = sys.argv[1:]
ns = arg_parser.parse_args(args, out=out, err=err)
conf = VmStack(ns.name)
vm = conf.get('vm.class')(conf)
if ns.download:
cmds.append(Download(vm))
if ns.ssh_keygen:
cmds.append(SshKeyGen(vm))
if ns.install:
cmds.append(Install(vm))
return cmds
def run(args=None):
cmds = build_commands(args)
for cmd in cmds:
try:
cmd.run()
except SetupVmError, e:
# Stop on first error
print 'ERROR: %s' % e
exit(-1)
if __name__ == "__main__":
run()
PK rI setup_vm/bin/__init__.pyPK sI^-
+ u1_test_utils-0.5.dist-info/DESCRIPTION.rstUNKNOWN
PK sIJ~ ~ ) u1_test_utils-0.5.dist-info/metadata.json{"extensions": {"python.details": {"contacts": [{"email": "ricardo.kirkner@canonical.com", "name": "Ricardo Kirkner", "role": "author"}], "document_names": {"description": "DESCRIPTION.rst"}}}, "generator": "bdist_wheel (0.26.0)", "metadata_version": "2.0", "name": "u1-test-utils", "summary": "Common testcases and utilities for testing Ubuntu One applications.", "version": "0.5"}PK sI+ ) u1_test_utils-0.5.dist-info/top_level.txtfabfile
setup_vm
u1testutils
PK sI''\ \ ! u1_test_utils-0.5.dist-info/WHEELWheel-Version: 1.0
Generator: bdist_wheel (0.26.0)
Root-Is-Purelib: true
Tag: py2-none-any
PK sIӥ $ u1_test_utils-0.5.dist-info/METADATAMetadata-Version: 2.0
Name: u1-test-utils
Version: 0.5
Summary: Common testcases and utilities for testing Ubuntu One applications.
Home-page: UNKNOWN
Author: Ricardo Kirkner
Author-email: ricardo.kirkner@canonical.com
License: UNKNOWN
Platform: UNKNOWN
UNKNOWN
PK sIV " u1_test_utils-0.5.dist-info/RECORDfabfile/__init__.py,sha256=r4UjhsecEFjyDll8MnoqT2WC9-M1W9tfV-GUx4uil2E,129
fabfile/development.py,sha256=0Mumcgomkc_uT-3HPhcDzI7zdmV-tKi455YPwand_M8,1623
fabfile/environment.py,sha256=KB23YOTm13qfwd5CnJE4a3a_1-_VleljU_-PwAHsw0c,3074
setup_vm/__init__.py,sha256=-JB1Ds5NCf_U4udnGf3k9OpnKoAqH-gKSPf17JTIp-Y,190
setup_vm/bin/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
setup_vm/bin/setup_vm.py,sha256=Q4ayy6Z3SNHFMPCHAupqluyy5qn_lDZX3Y0ilRDKiy8,51641
setup_vm/tests/__init__.py,sha256=hKXdu1IhjcDsh5mssrYwUoHmrSSw20mFlFntoZGk0aQ,5669
setup_vm/tests/test_code_conformance.py,sha256=HPwPWL8Vm_xv4DEJ4Z21RUessoLtfQCom-mNlB3D2ns,958
setup_vm/tests/test_setup_vm.py,sha256=4lbFwiHQzeKb14T0gnL9wbb5_lCvelIPb54kBwI92ZQ,41555
setup_vm/tests/test_test.py,sha256=784EVzxRIhOvMFK3TDijkuozoWnCMwLL1SJqUsWQc9I,1651
u1_test_utils-0.5.dist-info/DESCRIPTION.rst,sha256=OCTuuN6LcWulhHS3d5rfjdsQtW22n7HENFRh6jC6ego,10
u1_test_utils-0.5.dist-info/METADATA,sha256=ImlvN-3HPEuInWVeO1yLHqzCi0hYsqoLWcLStcw-hpU,265
u1_test_utils-0.5.dist-info/RECORD,,
u1_test_utils-0.5.dist-info/WHEEL,sha256=JTb7YztR8fkPg6aSjc571Q4eiVHCwmUDlX8PhuuqIIE,92
u1_test_utils-0.5.dist-info/metadata.json,sha256=ffVK5lfSlw7kSS3iAQLqKy0kKtO3bTGON3OhjVs-8Ac,382
u1_test_utils-0.5.dist-info/top_level.txt,sha256=Y9fnJJeGmSpL6R1h3pH4vybVePkNpf_16pASQ6xVlZc,29
u1testutils/__init__.py,sha256=4FdYq1QhEnfseFtVXCMw_8wLQbaC3UXVBIhzGsP6C-8,20
u1testutils/django.py,sha256=d4IAf-21a8NPzeEDJnpoH0HdiB4mw6j9q_hnei4Gka4,3348
u1testutils/logging.py,sha256=XYqDSUz6LjDUQmMLBaFZ-cCoMD4kMGg33O5ril-GMq8,2300
u1testutils/mail.py,sha256=f1NEea04_30GmL4a93oGF5AUc3TkZuxGYyF-naAhFLU,5187
u1testutils/wsgi_intercept.py,sha256=tePueookm69MdP9gou2GmyhaZq5f5yL_jkAYT7rFmIY,4615
u1testutils/pay/__init__.py,sha256=zLsYZmJJo_vuaPCQqjooPp50iIY6IK0gcftPEZ5fnyU,765
u1testutils/pay/api.py,sha256=ctusUOUCkVY61f8xvkiBloDaVUStNPFMNnxn3nJ3AFU,2900
u1testutils/pay/data.py,sha256=8RNZaI3VmeYklKR2AAOGjXWLDigKn_ZU0UUxujUgh6Q,2724
u1testutils/pay/environment.py,sha256=8piJcNptLjhWjq8I3uTW_EsZohfSjcCIBBy3SQ3Y0GQ,827
u1testutils/pay/selftests/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/pay/selftests/unit/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/pay/selftests/unit/test_pay_api_client.py,sha256=IKzWJ-h9qClJ84WWzPg37ibJF6P1atp5Bm79NBMa3Ek,3724
u1testutils/selftests/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/selftests/django_project/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/selftests/django_project/settings.py,sha256=fwH3UY324lao4S1FdUJhj9lhI9DNbQHGrjibQ1XM05Y,1046
u1testutils/selftests/static/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/selftests/static/test_static_code_analysis.py,sha256=pVxo98RCWHtqau1JCQSZjqnn8xaRoArsePDpIgohGlk,1021
u1testutils/selftests/unit/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/selftests/unit/test_django.py,sha256=JBER-qrPOosKrPyjzdvcCXgMrAM-p9BtrreJ7CO_ZhU,3512
u1testutils/selftests/unit/test_logging.py,sha256=oDsu5mn8quoX_gwnw0eVNrMwNcL6Z1mOYsynLr7h0Z0,986
u1testutils/selftests/unit/test_mail.py,sha256=4CQZ4Bzi9bgNnFxZOGwxTf_ay9f1iRdQUGyN6HoqjKM,7676
u1testutils/sso/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sso/api.py,sha256=qoox0M-bHA1M2nMbZYMOigdbiaojNydDvuKNxbeyLLk,1976
u1testutils/sso/data.py,sha256=FtK5qHitMjfDa0L1-sJcs1EzbM1MIYDNgH7iKBS7eKY,3422
u1testutils/sso/environment.py,sha256=SEqmvC12snb8vTjYnm1XD3MrlODMCFyn8RNe8Rx4rK0,880
u1testutils/sso/mail.py,sha256=WsuhOgJmaucdED3Bj_JTMKaIE9Rov3ybPxhK1LlyfvA,3594
u1testutils/sso/selftests/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sso/selftests/acceptance/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sso/selftests/acceptance/test_sst_helpers.py,sha256=wptvvy9ecoYESWbHmGnUdblzcUNMnFjtoNvjDg2N0gk,11032
u1testutils/sso/selftests/unit/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sso/selftests/unit/test_data.py,sha256=sxqACZbo0BPs6ihBocgBdm2SckjurlhZscyYuwUJjTg,2953
u1testutils/sso/selftests/unit/test_environment.py,sha256=E_BYThuvGaESTkMjMA7B6OxF22cx3BRF55dLwB27a24,1574
u1testutils/sso/selftests/unit/test_sso_api_client.py,sha256=G2RDZjIrqWloZ3BfQ2l7r_wj_4KSaIuR0Xqt2GaAIRk,4317
u1testutils/sso/selftests/unit/test_sso_mail.py,sha256=rsZEHXFCz77Gkha4JAlIxXNZTG6w5MXjkLGOLZyz9UM,6307
u1testutils/sso/sst/__init__.py,sha256=_XW3MySGwK1h33z3Wx_NxAG0xxazFlantd8YmvwWhjY,4847
u1testutils/sso/sst/pages.py,sha256=WEuS3zhrUuOfeBTGeBosCikSoSVTnygCRtI_g5d7-ew,13547
u1testutils/sst/__init__.py,sha256=jp4V8C70pzxYUycZN4zmi8ybORPvUntknNmF3Xgxv8w,6586
u1testutils/sst/config.py,sha256=0FY0WoFevFdhJjfYVO00usY4R3tT0OWVZkiNmVQUssE,420
u1testutils/sst/selftests/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sst/selftests/acceptance/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sst/selftests/acceptance/test_page.py,sha256=c7IPTw8w5W0LHGcxxu2LOZYpMdmccOClJtlvPjK3H7M,8069
u1testutils/sst/selftests/acceptance/test_string_page.py,sha256=UQ8YTBh5qNqIpxqKW8Ygu3iP3pKNPpNL2QLrZBEvtdw,1876
u1testutils/sst/selftests/unit/__init__.py,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
u1testutils/sst/selftests/unit/test_pages.py,sha256=It7QPPfkxNL-LCTYOdFx4VQW8X4gVJ1-QWgGFpDBD9I,6960
u1testutils/static/__init__.py,sha256=XYUeFYmU6sySA_t8-KTOYXJrpcxsZBx9-JBHhi92Seg,86
u1testutils/static/test_pep8_conformance.py,sha256=PILjFHSRdAbVpMdk_e7ADL8XjA2uskpEm7F4d1dqxuo,2516
u1testutils/static/test_pyflakes_analysis.py,sha256=wr97R2CTag7y1f3s8s4EYXyPQ1c5G6Uel-Vjf7T49_E,2850
PK rIm fabfile/__init__.pyPK rIWW W fabfile/development.pyPK rI]z = fabfile/environment.pyPK rI-eW s u1testutils/logging.pyPK rIcC C u1testutils/mail.pyPK rI6zm 1 u1testutils/__init__.pyPK rI1 `1 u1testutils/wsgi_intercept.pyPK rIa۞
C u1testutils/django.pyPK rIѤ P u1testutils/sst/config.pyPK rIb8 R u1testutils/sst/__init__.pyPK rI % l u1testutils/sst/selftests/__init__.pyPK rI 0 l u1testutils/sst/selftests/acceptance/__init__.pyPK rI'wT T 8 Hm u1testutils/sst/selftests/acceptance/test_string_page.pyPK rI3VQ 1 t u1testutils/sst/selftests/acceptance/test_page.pyPK rI * Ɣ u1testutils/sst/selftests/unit/__init__.pyPK rI-E[0 0 , u1testutils/sst/selftests/unit/test_pages.pyPK rIn8uT T u1testutils/pay/api.pyPK rI@ u1testutils/pay/__init__.pyPK rI[; ; F u1testutils/pay/environment.pyPK rId0¤
u1testutils/pay/data.pyPK rI % u1testutils/pay/selftests/__init__.pyPK rI * u1testutils/pay/selftests/unit/__init__.pyPK rIď 5 ! u1testutils/pay/selftests/unit/test_pay_api_client.pyPK rI ! u1testutils/selftests/__init__.pyPK rI
) ? u1testutils/selftests/unit/test_django.pyPK rIyH^ * > u1testutils/selftests/unit/test_logging.pyPK rI ' ` u1testutils/selftests/unit/test_mail.pyPK rI &
u1testutils/selftests/unit/__init__.pyPK rI (
u1testutils/selftests/static/__init__.pyPK rI l) 9 + u1testutils/selftests/static/test_static_code_analysis.pyPK rI 0 u1testutils/selftests/django_project/__init__.pyPK rIX9 0 u1testutils/selftests/django_project/settings.pyPK rIk
1 u1testutils/sso/mail.pyPK rIM g p% u1testutils/sso/api.pyPK rI \- u1testutils/sso/__init__.pyPK rIyp p - u1testutils/sso/environment.pyPK rIZٌp^
^
A1 u1testutils/sso/data.pyPK rI 4 4 > u1testutils/sso/sst/pages.pyPK rI W s u1testutils/sso/sst/__init__.pyPK rI % % u1testutils/sso/selftests/__init__.pyPK rI 0 h u1testutils/sso/selftests/acceptance/__init__.pyPK rI\+ + 8 u1testutils/sso/selftests/acceptance/test_sst_helpers.pyPK rI_ / $ u1testutils/sso/selftests/unit/test_sso_mail.pyPK rI6G* + u1testutils/sso/selftests/unit/test_data.pyPK rI * u1testutils/sso/selftests/unit/__init__.pyPK rIG 5 . u1testutils/sso/selftests/unit/test_sso_api_client.pyPK rI!]O& & 2 ^ u1testutils/sso/selftests/unit/test_environment.pyPK rI0n" " , u1testutils/static/test_pyflakes_analysis.pyPK rI=
"V V @ u1testutils/static/__init__.pyPK rI^ + u1testutils/static/test_pep8_conformance.pyPK rIo6I setup_vm/__init__.pyPK rIV ' setup_vm/tests/test_code_conformance.pyPK rIS^S S
setup_vm/tests/test_setup_vm.pyPK rIuƛ% % r setup_vm/tests/__init__.pyPK rI6Ys s setup_vm/tests/test_test.pyPK rIO6 { setup_vm/bin/setup_vm.pyPK rI j setup_vm/bin/__init__.pyPK sI^-
+ u1_test_utils-0.5.dist-info/DESCRIPTION.rstPK sIJ~ ~ ) u1_test_utils-0.5.dist-info/metadata.jsonPK sI+ ) u1_test_utils-0.5.dist-info/top_level.txtPK sI''\ \ ! u1_test_utils-0.5.dist-info/WHEELPK sIӥ $ u1_test_utils-0.5.dist-info/METADATAPK sIV " u1_test_utils-0.5.dist-info/RECORDPK ? ? ů