Lubuntu
/
britney2-ubuntu
mirror of https://git.launchpad.net/~ubuntu-release/britney/+git/britney2-ubuntu
8
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add a new policy to message bugs on SRU regressions

Browse Source 
It works like this. We wait until all tests have finished running. and
then grab their results. If there are any regressions, we mail each bug
with a link to pending-sru.html. There's a state file which records the
mails we've sent out, so that we don't mail the same bug multiple times.
wip/linux-policy
Łukasz 'sil2100' Zemczak 6 years ago committed by Iain Lane
parent 8c166cf5e1
commit 2559049b9d
3 changed files with 813 additions and 0 deletions
Show Stats Download Patch File Download Diff File

6
britney.py
Unescape View File

@ -221,6 +221,7 @@ from britney2.policies.policy import (AgePolicy,
) )
from britney2.policies.autopkgtest import AutopkgtestPolicy from britney2.policies.autopkgtest import AutopkgtestPolicy
from britney2.policies.sourceppa import SourcePPAPolicy from britney2.policies.sourceppa import SourcePPAPolicy
from britney2.policies.sruadtregression import SRUADTRegressionPolicy
from britney2.policies.email import EmailPolicy from britney2.policies.email import EmailPolicy
from britney2.policies.lpexcusebugs import LPExcuseBugsPolicy from britney2.policies.lpexcusebugs import LPExcuseBugsPolicy
from britney2.utils import (log_and_format_old_libraries, from britney2.utils import (log_and_format_old_libraries,
@ -533,6 +534,11 @@ class Britney(object):
self._policy_engine.add_policy(LinuxPolicy(self.options, self.suite_info)) self._policy_engine.add_policy(LinuxPolicy(self.options, self.suite_info))
self._policy_engine.add_policy(LPExcuseBugsPolicy(self.options, self.suite_info)) self._policy_engine.add_policy(LPExcuseBugsPolicy(self.options, self.suite_info))
self._policy_engine.add_policy(SourcePPAPolicy(self.options, self.suite_info)) self._policy_engine.add_policy(SourcePPAPolicy(self.options, self.suite_info))
add_sruregression_policy = getattr(self.options, 'sruregressionemail_enable', 'no')
if add_sruregression_policy in ('yes', 'dry-run'):
self._policy_engine.add_policy(SRUADTRegressionPolicy(self.options,
self.suite_info,
dry_run=add_sruregression_policy == 'dry-run'))
add_email_policy = getattr(self.options, 'email_enable', 'no') add_email_policy = getattr(self.options, 'email_enable', 'no')
if add_email_policy in ('yes', 'dry-run'): if add_email_policy in ('yes', 'dry-run'):
self._policy_engine.add_policy(EmailPolicy(self.options, self._policy_engine.add_policy(EmailPolicy(self.options,

218
britney2/policies/sruadtregression.py
Unescape View File

@ -0,0 +1,218 @@
import os
import json
import socket
import smtplib
from collections import defaultdict
from urllib.request import urlopen, URLError
from britney2 import SuiteClass
from britney2.policies.rest import Rest
from britney2.policies.policy import BasePolicy, PolicyVerdict
MESSAGE = """From: Ubuntu Stable Release Team <ubuntu-sru-bot@canonical.com>
To: {bug_mail}
Subject: Autopkgtest regression report ({source_name}/{version})
All autopkgtests for the newly accepted {source_name} ({version}) for {series_name} have finished running.
The following regressions have been reported in tests triggered by the package:
{failures}
Please visit the excuses page listed below and investigate the failures, proceeding afterwards as per the StableReleaseUpdates policy regarding autopkgtest regressions [1].
https://people.canonical.com/~ubuntu-archive/proposed-migration/{series_name}/update_excuses.html#{source_name}
[1] https://wiki.ubuntu.com/StableReleaseUpdates#Autopkgtest_Regressions
Thank you!
""" # noqa:E501
class SRUADTRegressionPolicy(BasePolicy, Rest):
def __init__(self, options, suite_info, dry_run=False):
super().__init__(
"sruadtregression",
options,
suite_info,
{SuiteClass.PRIMARY_SOURCE_SUITE},
)
self.state_filename = os.path.join(
options.unstable, "sru_regress_inform_state"
)
self.state = {}
self.dry_run = dry_run
self.email_host = getattr(self.options, "email_host", "localhost")
def initialise(self, britney):
super().initialise(britney)
if os.path.exists(self.state_filename):
with open(self.state_filename, encoding="utf-8") as data:
self.state = json.load(data)
self.logger.info("Loaded state file %s" % self.state_filename)
# Remove any old entries from the statefile
self.cleanup_state()
def bugs_from_changes(self, change_url):
"""Return bug list from a .changes file URL"""
last_exception = None
# Querying LP can timeout a lot, retry 3 times
for i in range(3):
try:
changes = urlopen(change_url)
break
except URLError as e:
last_exception = e
pass
else:
raise last_exception
bugs = set()
for ln in changes:
ln = ln.decode("utf-8")
if ln.startswith("Launchpad-Bugs-Fixed: "):
bugs = {int(b) for b in ln.split()[1:]}
break
return bugs
def apply_src_policy_impl(
self, policy_info, item, source_data_tdist, source_data_srcdist, excuse
):
source_name = item.package
# We only care about autopkgtest regressions
if (
"autopkgtest" not in excuse.reason
or not excuse.reason["autopkgtest"]
):
return PolicyVerdict.PASS
# Check the policy_info to see if all tests finished and, if yes, if we
# have any failures to report on.
if "autopkgtest" not in excuse.policy_info:
return PolicyVerdict.PASS
regressions = defaultdict(list)
for pkg in excuse.policy_info["autopkgtest"]:
# Unfortunately the verdict of the policy lives at the same level
# as its detailed output
if pkg == "verdict":
continue
for arch in excuse.policy_info["autopkgtest"][pkg]:
if excuse.policy_info["autopkgtest"][pkg][arch][0] == "RUNNING":
# If there's at least one test still running, we just stop
# and not proceed any further.
return PolicyVerdict.PASS
elif (
excuse.policy_info["autopkgtest"][pkg][arch][0]
== "REGRESSION"
):
regressions[pkg].append(arch)
if not regressions:
return PolicyVerdict.PASS
version = source_data_srcdist.version
distro_name = self.options.distribution
series_name = self.options.series
try:
if self.state[distro_name][series_name][source_name] == version:
# We already informed about the regression.
return PolicyVerdict.PASS
except KeyError:
# Expected when no failure has been reported so far for this
# source - we want to continue in that case. Easier than
# doing n number of if-checks.
pass
data = self.query_lp_rest_api(
"%s/+archive/primary" % distro_name,
{
"ws.op": "getPublishedSources",
"distro_series": "/%s/%s" % (distro_name, series_name),
"exact_match": "true",
"order_by_date": "true",
"pocket": "Proposed",
"source_name": source_name,
"version": version,
},
)
try:
src = next(iter(data["entries"]))
# IndexError means no packages in -proposed matched this name/version.
except StopIteration:
self.logger.info(
"No packages matching %s/%s the %s/%s main archive, not "
"informing of ADT regressions"
% (source_name, version, distro_name, series_name)
)
return PolicyVerdict.PASS
changes_url = self.query_lp_rest_api(
src["self_link"], {"ws.op": "changesFileUrl"}
)
if not changes_url:
return PolicyVerdict.PASS
# Prepare a helper string that lists all the ADT failures
failures = ""
for pkg, arches in regressions.items():
failures += "%s (%s)\n" % (pkg, ", ".join(arches))
bugs = self.bugs_from_changes(changes_url)
# Now leave a comment informing about the ADT regressions on each bug
for bug in bugs:
self.logger.info(
"%sSending ADT regression message to LP: #%s "
"regarding %s/%s in %s"
% (
"[dry-run] " if self.dry_run else "",
bug,
source_name,
version,
series_name,
)
)
if not self.dry_run:
try:
bug_mail = "%s@bugs.launchpad.net" % bug
server = smtplib.SMTP(self.email_host)
server.sendmail(
"noreply@canonical.com",
bug_mail,
MESSAGE.format(**locals()),
)
server.quit()
except socket.error as err:
self.logger.info(
"Failed to send mail! Is SMTP server running?"
)
self.logger.info(err)
self.save_progress(source_name, version, distro_name, series_name)
return PolicyVerdict.PASS
def save_progress(self, source, version, distro, series):
if self.dry_run:
return
if distro not in self.state:
self.state[distro] = {}
if series not in self.state[distro]:
self.state[distro][series] = {}
self.state[distro][series][source] = version
with open(self.state_filename, "w", encoding="utf-8") as data:
json.dump(self.state, data)
def cleanup_state(self):
"""Remove all no-longer-valid package entries from the statefile"""
for distro_name in self.state:
for series_name, pkgs in self.state[distro_name].items():
for source_name, version in pkgs.copy().items():
data = self.query_lp_rest_api(
"%s/+archive/primary" % distro_name,
{
"ws.op": "getPublishedSources",
"distro_series": "/%s/%s"
% (distro_name, series_name),
"exact_match": "true",
"order_by_date": "true",
"pocket": "Proposed",
"status": "Published",
"source_name": source_name,
"version": version,
},
)
if not data["entries"]:
del self.state[distro_name][series_name][source_name]

589
tests/test_sruadtregression.py
Unescape View File

@ -0,0 +1,589 @@
#!/usr/bin/python3
# (C) 2018 Canonical Ltd.
#
# 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 2 of the License, or
# (at your option) any later version.
import logging
import os
import sys
import json
import unittest
from tempfile import TemporaryDirectory
from unittest.mock import DEFAULT, Mock, patch, call
from urllib.request import URLError
PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, PROJECT_DIR)
from britney2.policies.policy import PolicyVerdict
from britney2.policies.sruadtregression import SRUADTRegressionPolicy
from tests.test_sourceppa import FakeBritney
FAKE_CHANGES = b"""Format: 1.8
Date: Mon, 16 Jul 2018 17:05:18 -0500
Source: test
Binary: test
Architecture: source
Version: 1.0
Distribution: bionic
Urgency: medium
Maintainer: Ubuntu Developers <ubuntu-devel-discuss@lists.ubuntu.com>
Changed-By: Foo Bar <foo.bar@ubuntu.com>
Description:
test - A test package
Launchpad-Bugs-Fixed: 1 4 2 31337 31337
Changes:
test (1.0) bionic; urgency=medium
.
* Test build
Checksums-Sha1:
fb11f859b85e09513d395a293dbb0808d61049a7 2454 test_1.0.dsc
06500cc627bc04a02b912a11417fca5a1109ec97 69852 test_1.0.debian.tar.xz
Checksums-Sha256:
d91f369a4b7fc4cba63540a81bd6f1492ca86661cf2e3ccac6028fb8c98d5ff5 2454 test_1.0.dsc
ffb460269ea2acb3db24adb557ba7541fe57fde0b10f6b6d58e8958b9a05b0b9 69852 test_1.0.debian.tar.xz
Files:
2749eba6cae4e49c00f25e870b228871 2454 libs extra test_1.0.dsc
95bf4af1ba0766b6f83dd3c3a33b0272 69852 libs extra test_1.0.debian.tar.xz
"""
class FakeOptions:
distribution = "testbuntu"
series = "zazzy"
unstable = "/tmp"
verbose = False
email_host = "localhost:1337"
class FakeSourceData:
version = "55.0"
class TestPackage:
package = "testpackage"
class FakeExcuse:
is_valid = True
daysold = 0
reason = {"autopkgtest": 1}
current_policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY
policy_info = {
"autopkgtest": {
"testpackage/55.0": {
"amd64": ["PASS", None, None, None, None],
"i386": ["REGRESSION", None, None, None, None],
},
"rdep1/1.1-0ubuntu1": {
"amd64": ["REGRESSION", None, None, None, None],
"i386": ["REGRESSION", None, None, None, None],
},
"rdep2/0.1-0ubuntu2": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
}
}
class FakeExcuseRunning:
is_valid = True
daysold = 0
reason = {"autopkgtest": 1}
current_policy_verdict = PolicyVerdict.REJECTED_TEMPORARILY
policy_info = {
"autopkgtest": {
"testpackage/55.0": {
"amd64": ["PASS", None, None, None, None],
"i386": ["REGRESSION", None, None, None, None],
},
"rdep1/1.1-0ubuntu1": {
"amd64": ["RUNNING", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
"rdep2/0.1-0ubuntu2": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
}
}
class FakeExcusePass:
is_valid = True
daysold = 0
reason = {}
current_policy_verdict = PolicyVerdict.PASS
policy_info = {
"autopkgtest": {
"testpackage/55.0": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
"rdep1/1.1-0ubuntu1": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
"rdep2/0.1-0ubuntu2": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
}
}
class FakeExcuseHinted:
is_valid = True
daysold = 0
reason = {"skiptest": 1}
current_policy_verdict = PolicyVerdict.PASS_HINTED
policy_info = {
"autopkgtest": {
"testpackage/55.0": {
"amd64": ["PASS", None, None, None, None],
"i386": ["ALWAYSFAIL", None, None, None, None],
},
"rdep1/1.1-0ubuntu1": {
"amd64": ["IGNORE-FAIL", None, None, None, None],
"i386": ["IGNORE-FAIL", None, None, None, None],
},
"rdep2/0.1-0ubuntu2": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
}
}
class FakeExcuseVerdictOverriden:
is_valid = True
daysold = 0
reason = {"skiptest": 1}
current_policy_verdict = PolicyVerdict.REJECTED_PERMANENTLY
policy_info = {
"autopkgtest": {
"testpackage/55.0": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
"rdep1/1.1-0ubuntu1": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
"rdep2/0.1-0ubuntu2": {
"amd64": ["PASS", None, None, None, None],
"i386": ["PASS", None, None, None, None],
},
}
}
class T(unittest.TestCase):
def setUp(self):
super().setUp()
@patch(
"britney2.policies.sruadtregression.urlopen",
return_value=iter(FAKE_CHANGES.split(b"\n")),
)
def test_bugs_from_changes(self, urlopen_mock):
"""Check extraction of bug numbers from .changes files"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pol = SRUADTRegressionPolicy(options, {})
bugs = pol.bugs_from_changes("http://some.url")
self.assertEqual(len(bugs), 4)
self.assertSetEqual(bugs, set((1, 4, 2, 31337)))
@patch(
"britney2.policies.sruadtregression.urlopen",
side_effect=URLError("timeout"),
)
def test_bugs_from_changes_retry(self, urlopen_mock):
"""Check .changes extraction retry mechanism"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pol = SRUADTRegressionPolicy(options, {})
self.assertRaises(
URLError, pol.bugs_from_changes, "http://some.url"
)
self.assertEqual(urlopen_mock.call_count, 3)
@patch.object(logging.Logger, "info")
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_comment_on_regression_and_update_state(
self, lp, bugs_from_changes, smtp, log
):
"""Verify bug commenting about ADT regressions and save the state"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = {}
pkg_mock[
"self_link"
] = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
lp.return_value = {"entries": [pkg_mock]}
previous_state = {
"testbuntu": {
"zazzy": {"testpackage": "54.0", "ignored": "0.1"}
},
"ghostdistro": {"spooky": {"ignored": "0.1"}},
}
excuse = FakeExcuse
pol = SRUADTRegressionPolicy(options, {})
# Set a base state
pol.state = previous_state
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
# Assert that we were looking for the right package as per
# FakeSourceData contents
self.assertSequenceEqual(
lp.mock_calls,
[
call(
"testbuntu/+archive/primary",
{
"distro_series": "/testbuntu/zazzy",
"exact_match": "true",
"order_by_date": "true",
"pocket": "Proposed",
"source_name": "testpackage",
"version": "55.0",
"ws.op": "getPublishedSources",
},
),
call(
"https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565",
{"ws.op": "changesFileUrl"},
),
],
)
# The .changes file only lists 2 bugs, make sure only those are
# commented on
self.assertSequenceEqual(
smtp.call_args_list,
[call("localhost:1337"), call("localhost:1337")],
)
self.assertEqual(smtp().sendmail.call_count, 2)
# call_args is a tuple (args, kwargs), and the email content is
# the third non-named argument of sendmail() - hence [0][2]
message = smtp().sendmail.call_args[0][2]
# Check if the e-mail/comment we were sending includes info about
# which tests have failed
self.assertIn("testpackage/55.0 (i386)", message)
self.assertIn("rdep1/1.1-0ubuntu1 (amd64, i386)", message)
# Check if the state has been saved and not overwritten
expected_state = {
"testbuntu": {
"zazzy": {"testpackage": "55.0", "ignored": "0.1"}
},
"ghostdistro": {"spooky": {"ignored": "0.1"}},
}
self.assertDictEqual(pol.state, expected_state)
log.assert_called_with(
"Sending ADT regression message to LP: #2 regarding testpackage/55.0 in zazzy"
)
# But also test if the state has been correctly recorded to the
# state file
state_path = os.path.join(
options.unstable, "sru_regress_inform_state"
)
with open(state_path) as f:
saved_state = json.load(f)
self.assertDictEqual(saved_state, expected_state)
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_no_comment_if_running(self, lp, bugs_from_changes, smtp):
"""Don't comment if tests still running"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = Mock()
pkg_mock.self_link = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
lp.return_value = {"entries": [pkg_mock]}
excuse = FakeExcuseRunning
pol = SRUADTRegressionPolicy(options, {})
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
bugs_from_changes.assert_not_called()
lp.assert_not_called()
smtp.sendmail.assert_not_called()
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_no_comment_if_passed(self, lp, bugs_from_changes, smtp):
"""Don't comment if all tests passed"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = Mock()
pkg_mock.self_link = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
bugs_from_changes.return_value = {"entries": [pkg_mock]}
excuse = FakeExcusePass
pol = SRUADTRegressionPolicy(options, {})
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
bugs_from_changes.assert_not_called()
lp.assert_not_called()
smtp.sendmail.assert_not_called()
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_no_comment_if_hinted(self, lp, bugs_from_changes, smtp):
"""Don't comment if package has been hinted in"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = Mock()
pkg_mock.self_link = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
bugs_from_changes.return_value = {"entries": [pkg_mock]}
excuse = FakeExcuseHinted
pol = SRUADTRegressionPolicy(options, {})
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
bugs_from_changes.assert_not_called()
lp.assert_not_called()
smtp.sendmail.assert_not_called()
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_no_comment_if_commented(self, lp, bugs_from_changes, smtp):
"""Don't comment if package has been already commented on"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = Mock()
pkg_mock.self_link = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
bugs_from_changes.return_value = {"entries": [pkg_mock]}
previous_state = {
"testbuntu": {
"zazzy": {"testpackage": "55.0", "ignored": "0.1"}
}
}
excuse = FakeExcuse
pol = SRUADTRegressionPolicy(options, {})
# Set a base state
pol.state = previous_state
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
bugs_from_changes.assert_not_called()
lp.assert_not_called()
smtp.sendmail.assert_not_called()
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_no_comment_if_verdict_overriden(self, lp, bugs_from_changes, smtp):
"""Make sure the previous 'bug' of commenting on uploads that had no
failing tests (or had them still running) because of the
current_policy_verdict getting reset to REJECTED_PERMANENTLY
by some other policy does not happen anymore."""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = Mock()
pkg_mock.self_link = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
lp.return_value = {"entries": [pkg_mock]}
# This should no longer be a valid bug as we switched how we
# actually determine if a regression is present, but still
# - better to have an explicit test case for it.
excuse = FakeExcuseVerdictOverriden
pol = SRUADTRegressionPolicy(options, {})
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
bugs_from_changes.assert_not_called()
lp.assert_not_called()
smtp.sendmail.assert_not_called()
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_initialize(self, lp):
"""Check state load, old package cleanup and LP login"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock1 = Mock()
pkg_mock1.source_name = "testpackage"
pkg_mock2 = Mock()
pkg_mock2.source_name = "otherpackage"
# Since we want to be as accurate as possible, we return query
# results per what query has been performed.
def query_side_effect(link, query):
if query["source_name"] == "testpackage":
return {"entries": [pkg_mock1]}
elif query["source_name"] == "otherpackage":
return {"entries": [pkg_mock2]}
return {"entries": []}
lp.side_effect = query_side_effect
state = {
"testbuntu": {
"zazzy": {
"testpackage": "54.0",
"toremove": "0.1",
"otherpackage": "13ubuntu1",
}
}
}
# Prepare the state file
state_path = os.path.join(
options.unstable, "sru_regress_inform_state"
)
with open(state_path, "w") as f:
json.dump(state, f)
pol = SRUADTRegressionPolicy(options, {})
pol.initialise(FakeBritney())
# Check if the stale packages got purged and others not
expected_state = {
"testbuntu": {
"zazzy": {
"testpackage": "54.0",
"otherpackage": "13ubuntu1",
}
}
}
# Make sure the state file has been loaded correctly
self.assertDictEqual(pol.state, expected_state)
# Check if we logged in with the right LP credentials
self.assertEqual(pol.email_host, "localhost:1337")
@patch.object(logging.Logger, "info")
@patch("smtplib.SMTP")
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.bugs_from_changes",
return_value={1, 2},
)
@patch(
"britney2.policies.sruadtregression.SRUADTRegressionPolicy.query_lp_rest_api"
)
def test_no_comment_dry_run(self, lp, bugs_from_changes, smtp, log):
"""Verify bug commenting about ADT regressions and save the state"""
with TemporaryDirectory() as tmpdir:
options = FakeOptions
options.unstable = tmpdir
pkg_mock = {}
pkg_mock[
"self_link"
] = "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565"
lp.return_value = {"entries": [pkg_mock]}
previous_state = {
"testbuntu": {
"zazzy": {"testpackage": "54.0", "ignored": "0.1"}
},
"ghostdistro": {"spooky": {"ignored": "0.1"}},
}
excuse = FakeExcuse
pol = SRUADTRegressionPolicy(options, {}, dry_run=True)
# Set a base state
pol.state = previous_state
status = pol.apply_src_policy_impl(
None, TestPackage, None, FakeSourceData, excuse
)
self.assertEqual(status, PolicyVerdict.PASS)
# Assert that we were looking for the right package as per
# FakeSourceData contents
self.assertSequenceEqual(
lp.mock_calls,
[
call(
"testbuntu/+archive/primary",
{
"distro_series": "/testbuntu/zazzy",
"exact_match": "true",
"order_by_date": "true",
"pocket": "Proposed",
"source_name": "testpackage",
"version": "55.0",
"ws.op": "getPublishedSources",
},
),
call(
"https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/9870565",
{"ws.op": "changesFileUrl"},
),
],
)
# Nothing happened
smtp.assert_not_called()
smtp.sendmail.assert_not_called()
self.assertDictEqual(pol.state, previous_state)
log.assert_called_with(
"[dry-run] Sending ADT regression message to LP: #2 regarding testpackage/55.0 in zazzy"
)
if __name__ == "__main__":
unittest.main()
Write Preview
Loading…
Cancel
Save