Preliminary code to make autopkgtest policy download autopkgtest.db

Include test to assert britney failure if the database is unavailable
less-recipients
Steve Langasek 3 years ago
parent 3814d63666
commit af8890bc4b

@ -124,6 +124,7 @@ class AutopkgtestPolicy(BasePolicy):
self.pending_tests_file = os.path.join(self.state_dir, 'autopkgtest-pending.json')
self.testsuite_triggers = {}
self.result_in_baseline_cache = collections.defaultdict(dict)
self.database = os.path.join(self.state_dir, 'autopkgtest.db')
# results map: trigger -> src -> arch -> [passed, version, run_id, seen]
# - trigger is "source/version" of an unstable package that triggered
@ -140,6 +141,11 @@ class AutopkgtestPolicy(BasePolicy):
else:
self.results_cache_file = os.path.join(self.state_dir, 'autopkgtest-results.cache')
if hasattr(self.options,'adt_db_url') and self.options.adt_db_url:
if not self.fetch_db():
self.logger.error('No autopkgtest db present, exiting')
sys.exit(1)
try:
self.options.adt_ppas = self.options.adt_ppas.strip().split()
except AttributeError:
@ -157,6 +163,34 @@ class AutopkgtestPolicy(BasePolicy):
else:
self.logger.info("Ignoring ADT_ARCHES %s as it is not in architectures list", arch)
def fetch_db(self):
f = None
try:
f = self.download_retry(self.options.adt_db_url)
http_code = f.getcode()
# file:/// urls don't have the http niceties
if not http_code or http_code == 200:
new_file = self.database + '.new'
with open(new_file,'wb') as f_out:
while True:
data=f.read(2048*1024)
if not data:
break
f_out.write(data)
if http_code and os.path.getsize(new_file) != int(f.getheader('content-length')):
self.logger.info('Short read downloading autopkgtest results')
os.unlink(new_file)
else:
os.rename(new_file, self.database)
else:
self.logger.error('Failure to fetch autopkgtest results %s: HTTP code=%d', self.options.adt_db_url, f.getcode())
except IOError as e:
self.logger.error('Failure to fetch autopkgtest results %s: %s', self.options.adt_db_url, str(e))
finally:
if f is not None:
f.close()
return os.path.exists(self.database)
def register_hints(self, hint_parser):
hint_parser.register_hint_type('force-badtest', britney2.hints.split_into_one_hint_per_package)
hint_parser.register_hint_type('force-skiptest', britney2.hints.split_into_one_hint_per_package)

@ -242,6 +242,29 @@ class AT(TestAutopkgtestBase):
# Tests for generic packages
################################################################
def test_fail_on_missing_database(self):
'''Fails if autopkgtest.db is requested but not available'''
os.unlink(self.db_path)
self.data.add_default_packages(lightgreen=False)
britney_failed = 0
try:
self.run_it(
# uninstallable unstable version
[('lightgreen', {'Version': '1.1~beta', 'Depends': 'libc6 (>= 0.9), libgreen1 (>= 2)'}, 'autopkgtest')],
{'lightgreen': (False, {})},
{'lightgreen': [('old-version', '1'), ('new-version', '1.1~beta'),
('reason', 'depends'),
('excuses', 'uninstallable on arch amd64, not running autopkgtest there')
]
})[1]
except AssertionError as e:
britney_failed = 1
self.assertEqual(britney_failed, 1, "DB missing but britney succeeded")
def test_no_request_for_uninstallable(self):
'''Does not request a test for an uninstallable package'''

Loading…
Cancel
Save