3eddcfccd8
F401 'dnf' imported but unused F401 'imp' imported but unused F401 'os' imported but unused F401 'subprocess' imported but unused F401 'sys' imported but unused F401 'yum' imported but unused JIRA: COMPOSE-4108 Signed-off-by: Haibo Lin <hlin@redhat.com>
148 lines
4.9 KiB
Python
148 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from datetime import datetime
|
|
import json
|
|
import mock
|
|
|
|
try:
|
|
import unittest2 as unittest
|
|
except ImportError:
|
|
import unittest
|
|
|
|
from pungi.notifier import PungiNotifier
|
|
|
|
|
|
mock_datetime = mock.Mock()
|
|
mock_datetime.utcnow.return_value = datetime(2017, 6, 28, 9, 34)
|
|
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
|
|
|
|
|
|
@mock.patch("pungi.util.makedirs")
|
|
@mock.patch("pungi.notifier.datetime", new=mock_datetime)
|
|
class TestNotifier(unittest.TestCase):
|
|
def setUp(self):
|
|
super(TestNotifier, self).setUp()
|
|
self.logfile = "/logs/notifications/notification-2017-06-28_09-34-00.log"
|
|
self.compose = mock.Mock(
|
|
compose_id="COMPOSE_ID",
|
|
compose_date="20171031",
|
|
compose_respin=1,
|
|
compose_label="Updates-20171031.1021",
|
|
compose_type="production",
|
|
log_warning=mock.Mock(),
|
|
conf={
|
|
"release_name": "Layer",
|
|
"release_short": "L",
|
|
"release_version": "27",
|
|
"release_type": "updates",
|
|
"release_is_layered": True,
|
|
"base_product_name": "Base",
|
|
"base_product_short": "B",
|
|
"base_product_version": "1",
|
|
"base_product_type": "ga",
|
|
},
|
|
paths=mock.Mock(
|
|
compose=mock.Mock(topdir=mock.Mock(return_value="/a/b")),
|
|
log=mock.Mock(topdir=mock.Mock(return_value="/logs")),
|
|
),
|
|
)
|
|
self.data = {"foo": "bar", "baz": "quux"}
|
|
|
|
def _call(self, script, cmd, **kwargs):
|
|
data = self.data.copy()
|
|
data["compose_id"] = "COMPOSE_ID"
|
|
data["location"] = "/a/b"
|
|
data["compose_date"] = "20171031"
|
|
data["compose_type"] = "production"
|
|
data["compose_respin"] = 1
|
|
data["compose_label"] = "Updates-20171031.1021"
|
|
data["release_short"] = "L"
|
|
data["release_name"] = "Layer"
|
|
data["release_version"] = "27"
|
|
data["release_type"] = "updates"
|
|
data["release_is_layered"] = True
|
|
data["base_product_name"] = "Base"
|
|
data["base_product_version"] = "1"
|
|
data["base_product_short"] = "B"
|
|
data["base_product_type"] = "ga"
|
|
data.update(kwargs)
|
|
return mock.call(
|
|
(script, cmd),
|
|
stdin_data=json.dumps(data),
|
|
can_fail=True,
|
|
return_stdout=False,
|
|
workdir=self.compose.paths.compose.topdir.return_value,
|
|
universal_newlines=True,
|
|
show_cmd=True,
|
|
logfile=self.logfile,
|
|
)
|
|
|
|
@mock.patch("pungi.util.translate_path")
|
|
@mock.patch("kobo.shortcuts.run")
|
|
def test_invokes_script(self, run, translate_path, makedirs):
|
|
run.return_value = (0, None)
|
|
translate_path.side_effect = lambda compose, x: x
|
|
|
|
n = PungiNotifier(["run-notify"])
|
|
n.compose = self.compose
|
|
n.send("cmd", **self.data)
|
|
|
|
makedirs.assert_called_once_with("/logs/notifications")
|
|
self.assertEqual(run.call_args_list, [self._call("run-notify", "cmd")])
|
|
|
|
@mock.patch("pungi.util.translate_path")
|
|
@mock.patch("kobo.shortcuts.run")
|
|
def test_invokes_multiple_scripts(self, run, translate_path, makedirs):
|
|
run.return_value = (0, None)
|
|
translate_path.side_effect = lambda compose, x: x
|
|
|
|
n = PungiNotifier(["run-notify", "ping-user"])
|
|
n.compose = self.compose
|
|
n.send("cmd", **self.data)
|
|
|
|
self.assertEqual(
|
|
sorted(run.call_args_list),
|
|
sorted([self._call("run-notify", "cmd"), self._call("ping-user", "cmd")]),
|
|
)
|
|
|
|
@mock.patch("kobo.shortcuts.run")
|
|
def test_translates_path(self, run, makedirs):
|
|
self.compose.paths.compose.topdir.return_value = "/root/a/b"
|
|
self.compose.conf["translate_paths"] = [
|
|
("/root/", "http://example.com/compose/")
|
|
]
|
|
|
|
run.return_value = (0, None)
|
|
|
|
n = PungiNotifier(["run-notify"])
|
|
n.compose = self.compose
|
|
n.send("cmd", **self.data)
|
|
|
|
self.assertEqual(
|
|
run.call_args_list,
|
|
[
|
|
self._call(
|
|
"run-notify", "cmd", location="http://example.com/compose/a/b"
|
|
)
|
|
],
|
|
)
|
|
|
|
@mock.patch("kobo.shortcuts.run")
|
|
def test_does_not_run_without_config(self, run, makedirs):
|
|
n = PungiNotifier(None)
|
|
n.send("cmd", foo="bar", baz="quux")
|
|
self.assertFalse(run.called)
|
|
|
|
@mock.patch("pungi.util.translate_path")
|
|
@mock.patch("kobo.shortcuts.run")
|
|
def test_logs_warning_on_failure(self, run, translate_path, makedirs):
|
|
translate_path.side_effect = lambda compose, x: x
|
|
run.return_value = (1, None)
|
|
|
|
n = PungiNotifier(["run-notify"])
|
|
n.compose = self.compose
|
|
n.send("cmd", **self.data)
|
|
|
|
self.assertEqual(run.call_args_list, [self._call("run-notify", "cmd")])
|
|
self.assertTrue(self.compose.log_warning.called)
|