Browse Source

autopkgtest: retrigger old failures

devuan-deployment
Paul Gevers 2 years ago
parent
commit
992b27abb3
No known key found for this signature in database GPG Key ID: 9C5C99EB05BD750A
6 changed files with 256 additions and 57 deletions
  1. +1
    -0
      britney.conf
  2. +134
    -37
      britney2/policies/autopkgtest.py
  3. +1
    -0
      tests/__init__.py
  4. +8
    -0
      tests/policy-test-data/autopkgtest/fail-old-test-result/debci.json
  5. +7
    -0
      tests/policy-test-data/autopkgtest/neutral-to-fail-pending-retest/debci.json
  6. +105
    -20
      tests/test_policy.py

+ 1
- 0
britney.conf View File

@@ -101,3 +101,4 @@ ADT_REGRESSION_PENALTY =
ADT_SUCCESS_BOUNTY = 3
ADT_BASELINE = reference
ADT_RETRY_URL_MECH = run_id
ADT_RETRY_OLDER_THAN = 1

+ 134
- 37
britney2/policies/autopkgtest.py View File

@@ -44,6 +44,7 @@ class Result(Enum):
FAIL = 1
PASS = 2
NEUTRAL = 3
NONE = 4


EXCUSES_LABELS = {
@@ -54,11 +55,14 @@ EXCUSES_LABELS = {
"REGRESSION": '<span style="background:#ff6666">Regression</span>',
"IGNORE-FAIL": '<span style="background:#e5c545">Ignored failure</span>',
"RUNNING": '<span style="background:#99ddff">Test in progress</span>',
"RUNNING-REFERENCE": '<span style="background:#ff6666">Reference test in progress, but real test failed already</span>',
"RUNNING-ALWAYSFAIL": '<span style="background:#99ddff">Test in progress (will not be considered a regression)</span>',
}

REF_TRIG = 'migration-reference/0'

SECPERDAY = 24 * 60 * 60


def srchash(src):
'''archive hash prefix for source package'''
@@ -164,6 +168,13 @@ class AutopkgtestPolicy(BasePolicy):
result.append(self._now)
self.test_results = results
self.logger.info('Read previous results from %s', self.results_cache_file)

# The cache can contain results against versions of packages that
# are not in any suite anymore. Strip those out, as we don't want
# to use those results.
if self.options.adt_baseline == 'reference':
self.filter_results_for_old_versions()

else:
self.logger.info('%s does not exist, re-downloading all results from swift', self.results_cache_file)

@@ -234,6 +245,51 @@ class AutopkgtestPolicy(BasePolicy):
else:
raise RuntimeError('Unknown ADT_AMQP schema %s' % amqp_url.split(':', 1)[0])

def filter_results_for_old_versions(self):
'''Remove results for old versions from the cache'''

test_results = self.test_results
test_results_new = deepcopy(test_results)

for (trigger, trigger_data) in test_results.items():
for (src, results) in trigger_data.items():
for (arch, result) in results.items():
if not self.test_version_in_any_suite(src, result[1]):
del test_results_new[trigger][src][arch]
if len(test_results_new[trigger][src]) == 0:
del test_results_new[trigger][src]
if len(test_results_new[trigger]) == 0:
del test_results_new[trigger]

self.test_results = test_results_new

def test_version_in_any_suite(self, src, version):
'''Check if the mentioned version of src is found in a suite

To prevent regressions in the target suite, the result should be
from a test with the version of the package in either the source
suite or the target suite. The source suite is also valid,
because due to versioned test dependencies and Breaks/Conflicts
relations, regularly the version in the source suite is used
during testing.
'''

versions = set()
for suite in self.suite_info:
try:
srcinfo = suite.sources[src]
except KeyError:
continue
versions.add(srcinfo.version)

valid_version = False
for ver in versions:
if apt_pkg.version_compare(ver, version) == 0:
valid_version = True
break

return valid_version

def save_state(self, britney):
super().save_state(britney)

@@ -300,7 +356,7 @@ class AutopkgtestPolicy(BasePolicy):
r = {v[0] for v in arch_results.values()}
if 'REGRESSION' in r:
verdict = PolicyVerdict.REJECTED_PERMANENTLY
elif 'RUNNING' in r and verdict == PolicyVerdict.PASS:
elif ('RUNNING' in r or 'RUNNING-REFERENCE' in r) and verdict == PolicyVerdict.PASS:
verdict = PolicyVerdict.REJECTED_TEMPORARILY
# skip version if still running on all arches
if not r - {'RUNNING', 'RUNNING-ALWAYSFAIL'}:
@@ -810,9 +866,7 @@ class AutopkgtestPolicy(BasePolicy):
self.logger.info('-> does not match any pending request for %s/%s', src, arch)

def add_trigger_to_results(self, trigger, src, ver, arch, run_id, seen, status):
# If a test runs because of its own package (newer version), ensure
# that we got a new enough version; FIXME: this should be done more
# generically by matching against testpkg-versions
# Ensure that we got a new enough version
try:
(trigsrc, trigver) = trigger.split('/', 1)
except ValueError:
@@ -821,6 +875,12 @@ class AutopkgtestPolicy(BasePolicy):
if trigsrc == src and apt_pkg.version_compare(ver, trigver) < 0:
self.logger.error('test trigger %s, but run for older version %s, ignoring', trigger, ver)
return
if self.options.adt_baseline == 'reference' and \
not self.test_version_in_any_suite(src, ver):
self.logger.error(
"Ignoring result for source %s and trigger %s as the tested version %s isn't found in any suite",
src, trigger, ver)
return

result = self.test_results.setdefault(trigger, {}).setdefault(
src, {}).setdefault(arch, [Result.FAIL, None, '', 0])
@@ -868,32 +928,53 @@ class AutopkgtestPolicy(BasePolicy):
of src. If huge is true, then the request will be put into the -huge
instead of normal queue.

This will only be done if that test wasn't already requested in a
previous run (i. e. not already in self.pending_tests) or there already
is a result for it. This ensures to download current results for this
package before requesting any test.
'''
This will only be done if that test wasn't already requested in
a previous run (i. e. if it's not already in self.pending_tests)
or if there is already a fresh or a positive result for it. This
ensures to download current results for this package before
requesting any test.
'''
trigger = full_trigger.split()[0]
# Don't re-request if we already have a result
uses_swift = not self.options.adt_swift_url.startswith('file://')
try:
result = self.test_results[trigger][src][arch][0]
if self.options.adt_swift_url.startswith('file://'):
result = self.test_results[trigger][src][arch]
has_result = True
except KeyError:
has_result = False

if has_result:
result_state = result[0]
version = result[1]
baseline = self.result_in_baseline(src, arch)
if result_state == Result.FAIL and \
baseline[0] in {Result.PASS, Result.NEUTRAL} and \
self.options.adt_retry_older_than and \
result[3] + int(self.options.adt_retry_older_than) * SECPERDAY < self._now:
# We might want to retry this failure, so continue
pass
elif not uses_swift:
# We're done if we don't retrigger and we're not using swift
return
if result in [Result.PASS, Result.NEUTRAL]:
elif result_state in {Result.PASS, Result.NEUTRAL}:
self.logger.info('%s/%s triggered by %s already known', src, arch, trigger)
return

# Without swift we don't expect new results
if uses_swift:
self.logger.info('Checking for new results for failed %s/%s for trigger %s', src, arch, trigger)
raise KeyError # fall through
except KeyError:
# Without swift we don't expect new results
if not self.options.adt_swift_url.startswith('file://'):
self.fetch_swift_results(self.options.adt_swift_url, src, arch)
# do we have one now?
try:
self.test_results[trigger][src][arch]
return
except KeyError:
pass
self.fetch_swift_results(self.options.adt_swift_url, src, arch)
# do we have one now?
try:
self.test_results[trigger][src][arch]
return
except KeyError:
pass

self.request_test_if_not_queued(src, arch, trigger, full_trigger, huge=huge)

def request_test_if_not_queued(self, src, arch, trigger, full_trigger=None, huge=False):
if full_trigger is None:
full_trigger = trigger

# Don't re-request if it's already pending
arch_list = self.pending_tests.setdefault(trigger, {}).setdefault(src, [])
@@ -918,31 +999,34 @@ class AutopkgtestPolicy(BasePolicy):
except KeyError:
pass

result_reference = Result.FAIL
result_reference = [Result.NONE, None, '', 0]
if self.options.adt_baseline == 'reference':
try:
result_reference = self.test_results[REF_TRIG][src][arch][0]
result_reference = self.test_results[REF_TRIG][src][arch]
self.logger.info('Found result for src %s in reference: %s',
src, result_reference.name)
src, result_reference[0].name)
except KeyError:
self.logger.info('Found NO result for src %s in reference: %s',
src, result_reference.name)
src, result_reference[0].name)
pass
self.result_in_baseline_cache[src][arch] = deepcopy(result_reference)
return result_reference

result_ever = Result.FAIL
result_ever = [Result.FAIL, None, '', 0]
for srcmap in self.test_results.values():
try:
if srcmap[src][arch][0] != Result.FAIL:
result_ever = srcmap[src][arch][0]
if result_ever == Result.PASS:
result_ever = srcmap[src][arch]
# If we are not looking at a reference run, We don't really
# care about anything except the status, so we're done
# once we find a PASS.
if result_ever[0] == Result.PASS:
break
except KeyError:
pass

self.result_in_baseline_cache[src][arch] = deepcopy(result_ever)
self.logger.info('Result for src %s ever: %s', src, result_ever.name)
self.logger.info('Result for src %s ever: %s', src, result_ever[0].name)
return result_ever

def pkg_test_result(self, src, ver, arch, trigger):
@@ -952,7 +1036,7 @@ class AutopkgtestPolicy(BasePolicy):
EXCUSES_LABELS. run_id is None if the test is still running.
'''
# determine current test result status
baseline_result = self.result_in_baseline(src, arch)
baseline_result = self.result_in_baseline(src, arch)[0]

url = None
run_id = None
@@ -973,11 +1057,24 @@ class AutopkgtestPolicy(BasePolicy):

if baseline_result == Result.FAIL:
result = 'ALWAYSFAIL'
else:
if self.has_force_badtest(src, ver, arch):
result = 'IGNORE-FAIL'
elif self.has_force_badtest(src, ver, arch):
result = 'IGNORE-FAIL'
elif baseline_result == Result.NONE:
# Check if the autopkgtest exists in the target suite and request it
test_in_target = False
try:
srcinfo = self.suite_info.target_suite.sources[src]
if 'autopkgtest' in srcinfo.testsuite:
test_in_target = True
except KeyError:
pass
if test_in_target:
self.request_test_if_not_queued(src, arch, REF_TRIG)
result = 'RUNNING-REFERENCE'
else:
result = 'REGRESSION'
result = 'ALWAYSFAIL'
else:
result = 'REGRESSION'
else:
result = r[0].name



+ 1
- 0
tests/__init__.py View File

@@ -401,6 +401,7 @@ ADT_HUGE = 20
ADT_SUCCESS_BOUNTY =
ADT_REGRESSION_PENALTY =
ADT_BASELINE =
ADT_RETRY_OLDER_THAN =
''')
assert os.path.exists(self.britney)



+ 8
- 0
tests/policy-test-data/autopkgtest/fail-old-test-result/debci.json View File

@@ -0,0 +1,8 @@
{"results":
[
{"trigger": "migration-reference/0", "package": "broken", "arch": "amd64", "version": "1.0", "status": "pass", "run_id": "1", "updated_at": "2018-10-03T21:12:00.000Z"},
{"trigger": "broken/2.0", "package": "broken", "arch": "amd64", "version": "2.0", "status": "pass", "run_id": "2", "updated_at": "2018-10-03T21:12:00.000Z"},
{"trigger": "migration-reference/0", "package": "inter", "arch": "amd64", "version": "1.0", "status": "pass", "run_id": "3", "updated_at": "2018-10-03T21:12:00.000Z"},
{"trigger": "broken/2.0", "package": "inter", "arch": "amd64", "version": "0.9", "status": "pass", "run_id": "4", "updated_at": "2018-10-03T21:12:00.000Z"}
]
}

+ 7
- 0
tests/policy-test-data/autopkgtest/neutral-to-fail-pending-retest/debci.json View File

@@ -0,0 +1,7 @@
{"results":
[
{"trigger": "migration-reference/0", "package": "pkg", "arch": "amd64", "version": "2.0", "status": "neutral", "run_id": "1", "updated_at": "2018-10-03T21:12:00.000Z"},
{"trigger": "pkg/2.0", "package": "pkg", "arch": "amd64", "version": "2.0", "status": "fail", "run_id": "2", "updated_at": "2018-10-03T21:12:00.000Z"},
{"trigger": "pkg/2.0", "package": "pkg", "arch": "amd64", "version": "2.0", "status": null, "run_id": "3", "updated_at": "2018-10-03T21:13:00.000Z"}
]
}

+ 105
- 20
tests/test_policy.py View File

@@ -21,9 +21,17 @@ def initialize_policy(test_name, policy_class, *args, **kwargs):
debci_data = os.path.join(test_dir, 'debci.json')
target = 'testing'
hints = []
pkg_universe = None
inst_tester = None
if 'hints' in kwargs:
hints = kwargs['hints']
del kwargs['hints']
if 'pkg_universe' in kwargs:
pkg_universe = kwargs['pkg_universe']
del kwargs['pkg_universe']
if 'inst_tester' in kwargs:
inst_tester = kwargs['inst_tester']
del kwargs['inst_tester']
options = MockObject(
state_dir=test_dir,
verbose=0,
@@ -47,7 +55,13 @@ def initialize_policy(test_name, policy_class, *args, **kwargs):
policy = policy_class(options, suite_info, *args)
fake_britney = MockObject(log=lambda x, y='I': None)
hint_parser = HintParser(mi_factory)
if pkg_universe and inst_tester:
build_sources_from_universe_and_inst_tester(policy, pkg_universe, inst_tester)
policy.initialise(fake_britney)
if inst_tester:
policy.britney._inst_tester = inst_tester
if pkg_universe:
policy.britney.pkg_universe = pkg_universe
policy.register_hints(hint_parser)
hint_parser.parse_hints(TEST_HINTER, HINTS_ALL, 'test-%s' % test_name, hints)
policy.hints = hint_parser.hints
@@ -114,8 +128,6 @@ def apply_src_policy(policy, expected_verdict, src_name, *, suite='unstable', ta

def build_sources_from_universe_and_inst_tester(policy, pkg_universe, inst_tester, suite='unstable'):
suite_info = policy.suite_info
policy.britney._inst_tester = inst_tester
policy.britney.pkg_universe = pkg_universe
src_universe = {}
bin_universe = {}
src_source = {}
@@ -360,8 +372,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/pass-to-pass',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_amqp=self.amqp,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.PASS, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'PASS'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
@@ -375,8 +388,25 @@ class TestAutopkgtestPolicy(unittest.TestCase):
'autopkgtest/pass-to-fail',
AutopkgtestPolicy,
adt_amqp=self.amqp,
adt_retry_older_than=1)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_retry_older_than=1,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_PERMANENTLY, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'REGRESSION'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][2] == 'packages/' + src_name[0] + '/' + src_name + '/testing/amd64'
amqp = self.read_amqp()
assert amqp[0:-1] == 'debci-testing-amd64:' + src_name + ' {"triggers": ["' + src_name + '/2.0"]}'

def test_pass_to_fail_no_retrigger(self):
src_name = 'pkg'
policy = initialize_policy(
'autopkgtest/pass-to-fail',
AutopkgtestPolicy,
adt_amqp=self.amqp,
adt_retry_older_than='',
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_PERMANENTLY, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'REGRESSION'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
@@ -389,8 +419,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/pass-to-neutral',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_amqp=self.amqp,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.PASS, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'NEUTRAL'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
@@ -407,8 +438,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/new',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, new_universe, new_inst_tester)
adt_amqp=self.amqp,
pkg_universe=new_universe,
inst_tester=new_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.PASS, src_name)
assert autopkgtest_policy_info[src_name][ARCH][0] == 'RUNNING-ALWAYSFAIL'
assert autopkgtest_policy_info[src_name][ARCH][1] == 'status/pending'
@@ -421,8 +453,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/pass-to-new',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_amqp=self.amqp,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_TEMPORARILY, src_name)
assert autopkgtest_policy_info[src_name][ARCH][0] == 'RUNNING'
assert autopkgtest_policy_info[src_name][ARCH][1] == 'status/pending'
@@ -435,8 +468,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/fail-to-new',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_amqp=self.amqp,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.PASS, src_name)
assert autopkgtest_policy_info[src_name][ARCH][0] == 'RUNNING-ALWAYSFAIL'
assert autopkgtest_policy_info[src_name][ARCH][1] == 'status/pending'
@@ -449,8 +483,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/neutral-to-new',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_amqp=self.amqp,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_TEMPORARILY, src_name)
assert autopkgtest_policy_info[src_name][ARCH][0] == 'RUNNING'
assert autopkgtest_policy_info[src_name][ARCH][1] == 'status/pending'
@@ -464,8 +499,25 @@ class TestAutopkgtestPolicy(unittest.TestCase):
'autopkgtest/neutral-to-fail',
AutopkgtestPolicy,
adt_amqp=self.amqp,
adt_retry_older_than=1)
build_sources_from_universe_and_inst_tester(policy, simple_universe, simple_inst_tester)
adt_retry_older_than=1,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_PERMANENTLY, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'REGRESSION'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][2] == 'packages/' + src_name[0] + '/' + src_name + '/testing/amd64'
amqp = self.read_amqp()
assert amqp[0:-1] == 'debci-testing-amd64:' + src_name + ' {"triggers": ["' + src_name + '/2.0"]}'

def test_neutral_to_fail_pending_retest(self):
src_name = 'pkg'
policy = initialize_policy(
'autopkgtest/neutral-to-fail-pending-retest',
AutopkgtestPolicy,
adt_amqp=self.amqp,
adt_retry_older_than=1,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_PERMANENTLY, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'REGRESSION'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
@@ -478,8 +530,9 @@ class TestAutopkgtestPolicy(unittest.TestCase):
policy = initialize_policy(
'autopkgtest/pass-to-new-with-breaks',
AutopkgtestPolicy,
adt_amqp=self.amqp)
build_sources_from_universe_and_inst_tester(policy, breaks_universe, breaks_inst_tester)
adt_amqp=self.amqp,
pkg_universe=breaks_universe,
inst_tester=breaks_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_TEMPORARILY, src_name)
assert autopkgtest_policy_info[src_name][ARCH][0] == 'RUNNING'
assert autopkgtest_policy_info[src_name][ARCH][1] == 'status/pending'
@@ -487,6 +540,38 @@ class TestAutopkgtestPolicy(unittest.TestCase):
amqp = self.read_amqp()
assert amqp[0:-1] == 'debci-testing-amd64:' + src_name + ' {"triggers": ["' + src_name + '/2.0 broken/2.0"]}'

def test_fail_old_test_result(self):
src_name = 'broken'
policy = initialize_policy(
'autopkgtest/fail-old-test-result',
AutopkgtestPolicy,
adt_amqp=self.amqp,
pkg_universe=breaks_universe,
inst_tester=breaks_inst_tester,
adt_baseline='reference')
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.REJECTED_TEMPORARILY, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'PASS'
assert autopkgtest_policy_info['inter'][ARCH][0] == 'RUNNING'
assert autopkgtest_policy_info['inter'][ARCH][1] == 'status/pending'
amqp = self.read_amqp()
assert amqp[0:-1] == 'debci-testing-amd64:inter {"triggers": ["' + src_name + '/2.0"]}'

def test_fail_to_fail(self):
src_name = 'pkg'
policy = initialize_policy(
'autopkgtest/fail-to-fail',
AutopkgtestPolicy,
adt_amqp=self.amqp,
adt_retry_older_than=1,
pkg_universe=simple_universe,
inst_tester=simple_inst_tester)
autopkgtest_policy_info = apply_src_policy(policy, PolicyVerdict.PASS, src_name)
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][0] == 'ALWAYSFAIL'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][1] == 'data/autopkgtest/testing/amd64/' + src_name[0] + '/' + src_name + '/2/log.gz'
assert autopkgtest_policy_info[src_name + '/2.0'][ARCH][2] == 'packages/' + src_name[0] + '/' + src_name + '/testing/amd64'
amqp = self.read_amqp()
assert len(amqp) == 0


if __name__ == '__main__':
unittest.main()

Loading…
Cancel
Save