#!/usr/bin/env python # -*- coding: utf-8 -*- # debianbts_test.py - Unittests for debianbts.py. # Copyright (C) 2009 Bastian Venthur # # 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. # # 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, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. from __future__ import division, unicode_literals, absolute_import, print_function import datetime import email.message import math import random import unittest import logging import threading try: import unittest.mock as mock except ImportError: import mock from pysimplesoap.simplexml import SimpleXMLElement import debianbts as bts logger = logging.getLogger(__name__) logging.basicConfig(level=logging.NOTSET) class DebianBtsTestCase(unittest.TestCase): def setUp(self): self.b1 = bts.Bugreport() self.b1.severity = 'normal' self.b2 = bts.Bugreport() self.b2.severity = 'normal' def test_get_usertag_empty(self): """get_usertag should return empty dict if no bugs are found.""" d = bts.get_usertag("thisisatest@debian.org") self.assertEqual(d, dict()) def test_get_usertag(self): """get_usertag should return dict with tag(s) and buglist(s).""" d = bts.get_usertag("debian-python@lists.debian.org") self.assertEqual(type(d), type(dict())) for k, v in d.items(): self.assertEqual(type(""), type(k)) self.assertEqual(type([]), type([])) for bug in v: self.assertEqual(type(bug), int) def test_get_usertag_filters(self): """get_usertag should return only requested tags""" tags = bts.get_usertag("debian-python@lists.debian.org") self.assertTrue(isinstance(tags, dict)) randomKey0 = random.choice(list(tags.keys())) randomKey1 = random.choice(list(tags.keys())) filtered_tags = bts.get_usertag( "debian-python@lists.debian.org", randomKey0, randomKey1) self.assertEqual(len(filtered_tags), 2) self.assertEqual(set(filtered_tags[randomKey0]), set(tags[randomKey0])) self.assertEqual(set(filtered_tags[randomKey1]), set(tags[randomKey1])) def test_get_bugs_empty(self): """get_bugs should return empty list if no matching bugs where found.""" l = bts.get_bugs("package", "thisisatest") self.assertEqual(l, []) def test_get_bugs(self): """get_bugs should return list of bugnumbers.""" l = bts.get_bugs("submitter", "venthur@debian.org") self.assertFalse(len(l) == 0) self.assertEqual(type(l), type([])) for i in l: self.assertEqual(type(i), type(int())) def test_get_bugs_list(self): """previous versions of python-debianbts accepted malformed key-value lists.""" l = bts.get_bugs('submitter', 'venthur@debian.org', 'severity', 'normal') l2 = bts.get_bugs(['submitter', 'venthur@debian.org', 'severity', 'normal']) self.assertFalse(len(l) == 0) l.sort() l2.sort() self.assertEqual(l, l2) def test_newest_bugs(self): """newest_bugs should return list of bugnumbers.""" l = bts.newest_bugs(10) self.assertEqual(type(l), type([])) for i in l: self.assertEqual(type(i), type(int())) def test_newest_bugs_amount(self): """newest_bugs(amount) should return a list of len 'amount'. """ for i in 0, 1, 10: l = bts.newest_bugs(i) self.assertEqual(len(l), i) def test_get_bug_log(self): """get_bug_log should return the correct data types.""" bl = bts.get_bug_log(223344) self.assertEqual(type(bl), type([])) for i in bl: self.assertEqual(type(i), type(dict())) self.assertTrue("attachments" in i) self.assertEqual(type(i["attachments"]), type(list())) self.assertTrue("body" in i) self.assertTrue(isinstance(i["body"], type(''))) self.assertTrue("header" in i) self.assertTrue(isinstance(i["header"], type(''))) self.assertTrue("msg_num" in i) self.assertEqual(type(i["msg_num"]), type(int())) def test_get_bug_log_with_attachments(self): """get_bug_log should include attachments""" buglogs = bts.get_bug_log(400000) for bl in buglogs: self.assertTrue("attachments" in bl) def test_bug_log_message(self): """dict returned by get_bug_log has a email.Message field""" buglogs = bts.get_bug_log(400012) for buglog in buglogs: self.assertTrue('message' in buglog) msg = buglog['message'] self.assertIsInstance(msg, email.message.Message) self.assertTrue('Subject' in msg) if not msg.is_multipart(): self._assert_unicode(msg.get_payload()) def test_bug_log_message_unicode(self): """test parsing of bug_log mail with non ascii characters""" buglogs = bts.get_bug_log(773321) buglog = buglogs[2] msg_payload = buglog['message'].get_payload() self._assert_unicode(msg_payload) self.assertTrue('é' in msg_payload) def test_empty_get_status(self): """get_status should return empty list if bug doesn't exits""" bugs = bts.get_status(0) self.assertEqual(type(bugs), list) self.assertEqual(len(bugs), 0) def test_sample_get_status(self): """test retrieving of a "known" bug status""" bugs = bts.get_status(486212) self.assertEqual(len(bugs), 1) bug = bugs[0] self.assertEqual(bug.bug_num, 486212) self.assertEqual(bug.date, datetime.datetime(2008, 6, 14, 10, 30, 2)) self.assertTrue(bug.subject.startswith('[reportbug-ng] segm')) self.assertEqual(bug.package, 'reportbug-ng') self.assertEqual(bug.severity, 'normal') self.assertEqual(bug.tags, ['help']) self.assertEqual(bug.blockedby, []) self.assertEqual(bug.blocks, []) self.assertEqual(bug.summary, '') self.assertEqual(bug.location, 'archive') self.assertEqual(bug.source, 'reportbug-ng') self.assertEqual(bug.log_modified, datetime.datetime(2008, 8, 17, 7, 26, 22)) self.assertEqual(bug.pending, 'done') self.assertEqual(bug.done, True) self.assertEqual(bug.archived, True) self.assertEqual(bug.found_versions, ['reportbug-ng/0.2008.06.04']) self.assertEqual(bug.fixed_versions, ['reportbug-ng/1.0']) self.assertEqual(bug.affects, []) def test_bug_str(self): """test string conversion of a Bugreport""" self.b2.package = 'foo-pkg' self.b2.bug_num = 12222 s = str(self.b2) self.assertTrue(isinstance(s, str)) # byte string in py2, unicode in py3 self.assertTrue('bug_num: 12222\n' in s) self.assertTrue('package: foo-pkg\n' in s) def test_get_status_affects(self): """test a bug with "affects" field""" bugs = bts.get_status(290501, 770490) self.assertEqual(len(bugs), 2) self.assertEqual(bugs[0].affects, []) self.assertEqual(bugs[1].affects, ['conkeror']) def test_status_batches_large_bug_counts(self): """get_status should perform requests in batches to reduce server load.""" with mock.patch.object(bts, '_build_soap_client') as mock_build_client: mock_build_client.return_value = mock_client = mock.Mock() mock_client.call.return_value = SimpleXMLElement( '') nr = bts.BATCH_SIZE + 10.0 calls = int(math.ceil(nr / bts.BATCH_SIZE)) bts.get_status([722226] * int(nr)) self.assertEqual(mock_client.call.call_count, calls) def test_status_batches_multiple_arguments(self): """get_status should batch multiple arguments into one request.""" with mock.patch.object(bts, '_build_soap_client') as mock_build_client: mock_build_client.return_value = mock_client = mock.Mock() mock_client.call.return_value = SimpleXMLElement( '') batch_size = bts.BATCH_SIZE calls = 1 bts.get_status(*list(range(batch_size))) self.assertEqual(mock_client.call.call_count, calls) calls += 2 bts.get_status(*list(range(batch_size + 1))) self.assertEqual(mock_client.call.call_count, calls) def test_comparison(self): """comparison of two bugs""" self.b1.archived = True self.b2.done = True self.assertTrue(self.b2 > self.b1) self.assertTrue(self.b2 >= self.b1) self.assertFalse(self.b2 == self.b1) self.assertFalse(self.b2 < self.b1) self.assertFalse(self.b2 <= self.b1) def test_comparison_equal(self): """comparison of two bug which are equal regarding their relative order""" self.b1.done = True self.b2.done = True self.assertFalse(self.b2 > self.b1) self.assertTrue(self.b2 >= self.b1) self.assertTrue(self.b2 == self.b1) self.assertFalse(self.b2 < self.b1) self.assertTrue(self.b2 <= self.b1) def test_get_bugs_int_bugs(self): """It is possible to pass a list of bug number to get_bugs""" bugs = bts.get_bugs('bugs', [400010, 400012], 'archive', True) self.assertEquals(set(bugs), set((400010, 400012))) def test_get_bugs_single_int_bug(self): """bugs parameter in get_bugs can be a list of int or a int""" bugs1 = bts.get_bugs('bugs', 400040, 'archive', True) bugs2 = bts.get_bugs('bugs', [400040], 'archive', True) self.assertEquals(bugs1, bugs2) def test_mergedwith(self): """Mergedwith is always a list of int.""" # this one is merged with two other bugs m = bts.get_status(486212)[0].mergedwith self.assertTrue(len(m) == 2) for i in m: self.assertEqual(type(i), type(int())) # this one was merged with one bug m = bts.get_status(433550)[0].mergedwith self.assertTrue(len(m) == 1) self.assertEqual(type(m[0]), type(int())) # this one was not merged m = bts.get_status(474955)[0].mergedwith self.assertEqual(m, list()) def test_base64_status_fields(self): """fields in bug status are sometimes base64-encoded""" bug = bts.get_status(711111)[0] self._assert_unicode(bug.originator) self.assertTrue(bug.originator.endswith('gmail.com>')) self.assertTrue('ł' in bug.originator) def test_base64_buglog_body(self): """buglog body is sometimes base64 encoded""" buglog = bts.get_bug_log(773321) body = buglog[2]['body'] self._assert_unicode(buglog[1]['body']) self.assertTrue('é' in body) def test_string_status_originator(self): """test reading of bug status originator that is not base64-encoded""" bug = bts.get_status(711112)[0] self._assert_unicode(bug.originator) self.assertTrue(bug.originator.endswith('debian.org>')) def test_unicode_conversion_in_str(self): """string representation must deal with unicode correctly.""" [bug] = bts.get_status(773321) try: bug.__str__() except UnicodeEncodeError: self.fail() def test_regression_588954(self): """Get_bug_log must convert the body correctly to unicode.""" try: bts.get_bug_log(582010) except UnicodeDecodeError: self.fail() def test_regression_590073(self): """bug.blocks is sometimes a str sometimes an int.""" try: # test the int case # TODO: test the string case bts.get_status(568657) except TypeError: self.fail() def test_regression_590725(self): """bug.body utf sometimes contains invalid continuation bytes.""" try: bts.get_bug_log(578363) bts.get_bug_log(570825) except UnicodeDecodeError: self.fail() def test_regression_670446(self): """affects should be split by ','""" bug = bts.get_status(657408)[0] self.assertEqual( bug.affects, ['epiphany-browser-dev', 'libwebkit-dev']) def test_regression_799528(self): """fields of buglog are sometimes base64 encoded.""" # bug with base64 encoding originator [bug] = bts.get_status(711111) self.assertTrue('ł' in bug.originator) # bug with base64 encoding subject [bug] = bts.get_status(779005) self.assertTrue('‘' in bug.subject) def _assert_unicode(self, string): """asserts for type of a unicode string, depending on python version""" if bts.PY2: self.assertIsInstance(string, unicode) else: self.assertIsInstance(string, str) class ThreadingTestCase(unittest.TestCase): """this class tests that the module is thread safe""" def setUp(self): self._thread_failed = False def test_multithreading(self): threads = [ threading.Thread(target=self._get_bugs_thread, args=(pkg,)) for pkg in ('python3-gst-1.0', 'libsoxr0') ] + [ threading.Thread(target=self._get_bug_log_thread, args=(bug_n,)) for bug_n in (300000, 300001) ] for thread in threads: thread.start() for thread in threads: thread.join() if self._thread_failed: self.fail('multithreaded calls failed') def _get_bugs_thread(self, pkg): try: bugs = bts.get_bugs('package', pkg) except Exception as exc: self._thread_failed = True print('threaded get_bugs() call failed ' 'with exception {} {}'.format(type(exc), exc)) def _get_bug_log_thread(self, bug_num): try: bug_logs = bts.get_bug_log(bug_num) except Exception as exc: self._thread_failed = True print('threaded get_bug_log() call failed ' 'with exception {} {}'.format(type(exc), exc)) if __name__ == "__main__": unittest.main()