diff options
| -rw-r--r-- | yarnhelper.py | 148 | ||||
| -rw-r--r-- | yarnhelper_tests.py | 134 |
2 files changed, 0 insertions, 282 deletions
diff --git a/yarnhelper.py b/yarnhelper.py deleted file mode 100644 index e265d82..0000000 --- a/yarnhelper.py +++ /dev/null @@ -1,148 +0,0 @@ -# Copyright 2017 Lars Wirzenius -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU 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 <http://www.gnu.org/licenses/>. -# -# =*= License: GPL-3+ =*= - - -import email -import imaplib -import os -import subprocess -import sys -import urlparse - -import cliapp -import requests -import yaml - - -variables_filename = os.environ.get('VARIABLES', 'vars.yaml') - - -class YarnHelper(object): - - def __init__(self): - self._env = dict(os.environ) - self._next_match = 1 - self._variables = None # None means not loaded, otherwise dict - - def set_environment(self, env): - self._env = dict(env) - - def get_next_match(self): - name = 'MATCH_{}'.format(self._next_match) - if name not in self._env: - raise Error('no next match') - self._next_match += 1 - return self._env[name] - - def get_variable(self, name, default=None): - if self._variables is None: - self._variables = self._load_variables() - return self._variables.get(name, default) - - def _load_variables(self): - if os.path.exists(variables_filename): - with open(variables_filename, 'r') as f: - return yaml.safe_load(f) - return {} - - def set_variable(self, name, value): - if self._variables is None: - self._variables = {} - self._variables[name] = value - self._save_variables(self._variables) - - def _save_variables(self, variables): - with open(variables_filename, 'w') as f: - yaml.safe_dump(variables, f) - - def append_to_list(self, list_name, value): - if self._variables is None: - self._variables = self._load_variables() - items = self._variables.get(list_name, []) - items.append(value) - self.set_variable(list_name, items) - - def construct_aliased_http_request( - self, address, method, url, data=None, headers=None): - - if headers is None: - headers = {} - - parts = list(urlparse.urlparse(url)) - headers['Host'] = parts[1] - parts[1] = address - aliased_url = urlparse.urlunparse(parts) - - r = requests.Request(method, aliased_url, data=data, headers=headers) - return r.prepare() - - def http_get(self, address, url): # pragma: no cover - r = self.construct_aliased_http_request(address, 'GET', url) - s = requests.Session() - resp = s.send(r) - return resp.status_code, resp.content - - def assertEqual(self, a, b): - if a != b: - raise Error('assertion {!r} == {!r} failed'.format(a, b)) - - def assertNotEqual(self, a, b): - if a == b: - raise Error('assertion {!r} != {!r} failed'.format(a, b)) - - def assertGreaterThan(self, a, b): - if a <= b: - raise Error('assertion {!r} > {!r} failed'.format(a, b)) - - def assertIn(self, a, b): - if a not in b: - raise Error('assertion {!r} in {!r} failed'.format(a, b)) - - def get_password_with_pass(self, pass_home, pass_name): # pragma: no cover - p = subprocess.Popen( - ['env', 'HOME={}'.format(pass_home), 'pass', 'show', pass_name], - stdout=subprocess.PIPE) - stdout, stderr = p.communicate() - password = stdout.rstrip() - return password - - def iterate_mails_in_imap_mailbox( - self, address, user, password, callback, exp): # pragma: no cover - m = imaplib.IMAP4_SSL(address) - m.login(user, password) - m.select('INBOX', False) - typ, data = m.search(None, 'ALL') - for num in data[0].split(): - typ, data = m.fetch(num, '(RFC822)') - typ, text = data[0] - msg = email.message_from_string(text) - callback(m, num, msg) - if exp: - m.expunge() - m.close() - m.logout() - - def repo_ssh_url(self, repo): # pragma: no cover - return 'ssh://git@{}/{}'.format(os.environ['GITANO_SERVER'], repo) - - def local_checkout_dirname(self, user, repo): # pragma: no cover - return '{}_{}'.format(user, repo) - - -class Error(Exception): - - pass diff --git a/yarnhelper_tests.py b/yarnhelper_tests.py deleted file mode 100644 index e503dd7..0000000 --- a/yarnhelper_tests.py +++ /dev/null @@ -1,134 +0,0 @@ -# Copyright 2017 Lars Wirzenius -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU 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 <http://www.gnu.org/licenses/>. -# -# =*= License: GPL-3+ =*= - - -import os -import unittest - -import yarnhelper - - -class GetNextMatchTests(unittest.TestCase): - - def test_raises_error_if_no_next_match(self): - h = yarnhelper.YarnHelper() - h.set_environment({}) - with self.assertRaises(yarnhelper.Error): - h.get_next_match() - - def test_returns_first_match_if_there(self): - h = yarnhelper.YarnHelper() - h.set_environment({ - 'MATCH_1': 'first', - }) - self.assertEqual(h.get_next_match(), 'first') - - def test_returns_second_match_if_there(self): - h = yarnhelper.YarnHelper() - h.set_environment({ - 'MATCH_1': 'first', - 'MATCH_2': 'second', - }) - self.assertEqual(h.get_next_match(), 'first') - self.assertEqual(h.get_next_match(), 'second') - - def test_raises_error_if_no_more_matches(self): - h = yarnhelper.YarnHelper() - h.set_environment({ - 'MATCH_1': 'first', - }) - self.assertEqual(h.get_next_match(), 'first') - with self.assertRaises(yarnhelper.Error): - h.get_next_match() - - -class PersistentVariableTests(unittest.TestCase): - - def setUp(self): - # We need this so that tearDown works - pass - - def tearDown(self): - if os.path.exists(yarnhelper.variables_filename): - os.remove(yarnhelper.variables_filename) - - def test_sets_variable_persistently(self): - h = yarnhelper.YarnHelper() - h.set_variable('FOO', 'bar') - - h2 = yarnhelper.YarnHelper() - self.assertEqual(h2.get_variable('FOO'), 'bar') - - def test_get_returns_default_if_variable_not_set(self): - h = yarnhelper.YarnHelper() - self.assertEqual(h.get_variable('FOO', 'bar'), 'bar') - - def test_appends_to_empty_list(self): - h = yarnhelper.YarnHelper() - h.append_to_list('foo', 1) - self.assertEqual(h.get_variable('foo'), [1]) - - -class HttpTests(unittest.TestCase): - - def test_constructs_aliased_request(self): - h = yarnhelper.YarnHelper() - server = 'new.example.com' - url = 'http://www.example.com/path' - r = h.construct_aliased_http_request(server, 'GET', url) - self.assertEqual(r.url, 'http://new.example.com/path') - self.assertEqual(r.headers['Host'], 'www.example.com') - - -class AssertionTests(unittest.TestCase): - - def test_assertEqual_asserts_equals_correctly(self): - h = yarnhelper.YarnHelper() - self.assertEqual(h.assertEqual(0, 0), None) - - def test_assertEqual_raises_error_for_nonequal_values(self): - h = yarnhelper.YarnHelper() - with self.assertRaises(yarnhelper.Error): - h.assertEqual(0, 1) - - def test_assertNotEqual_asserts_nonequal_correct(self): - h = yarnhelper.YarnHelper() - self.assertEqual(h.assertNotEqual(0, 1), None) - - def test_assertNotEqual_raises_error_for_equal_values(self): - h = yarnhelper.YarnHelper() - with self.assertRaises(yarnhelper.Error): - h.assertNotEqual(0, 0) - - def test_assertGreaterThan_raises_error_for_equal_values(self): - h = yarnhelper.YarnHelper() - with self.assertRaises(yarnhelper.Error): - h.assertGreaterThan(0, 0) - - def test_assertGreaterThan_raises_error_for_unordered_values(self): - h = yarnhelper.YarnHelper() - with self.assertRaises(yarnhelper.Error): - h.assertGreaterThan(0, 1) - - def test_assertIn_asserts_correctly(self): - h = yarnhelper.YarnHelper() - self.assertEqual(h.assertIn('ana', 'banana'), None) - - def test_assertIn_raises_error_for_false(self): - h = yarnhelper.YarnHelper() - with self.assertRaises(yarnhelper.Error): - h.assertIn('nope', 'banana') |