# Path: bugwarrior/command.py
# def _get_section_name(flavor):
# def _try_load_config(main_section, interactive=False):
# def pull(dry_run, flavor, interactive, debug):
# def vault():
# def targets():
# def list():
# def clear(target, username):
# def set(target, username):
# def uda(flavor):
# 
# Path: bugwarrior/config/load.py
# class BugwarriorConfigParser(configparser.ConfigParser):
#     def __init__(self, *args, allow_no_value=True, **kwargs):
#         super().__init__(*args, allow_no_value=allow_no_value, **kwargs)
# 
#     def getint(self, section, option):
#         """ Accepts both integers and empty values. """
#         try:
#             return super().getint(section, option)
#         except ValueError:
#             if self.get(section, option) == '':
#                 return None
#             else:
#                 raise ValueError(
#                     "{section}.{option} must be an integer or empty.".format(
#                         section=section, option=option))
# 
#     @staticmethod
#     def optionxform(option):
#         """ Do not lowercase key names. """
#         return option
# 
# Path: tests/base.py
# class ConfigTest(unittest.TestCase):
#     """
#     Creates config files, configures the environment, and cleans up afterwards.
#     """
#     def setUp(self):
#         self.old_environ = os.environ.copy()
#         self.tempdir = tempfile.mkdtemp(prefix='bugwarrior')
# 
#         # Create temporary config files.
#         self.taskrc = os.path.join(self.tempdir, '.taskrc')
#         self.lists_path = os.path.join(self.tempdir, 'lists')
#         os.mkdir(self.lists_path)
#         with open(self.taskrc, 'w+') as fout:
#             fout.write('data.location=%s\n' % self.lists_path)
# 
#         # Configure environment.
#         os.environ['HOME'] = self.tempdir
#         os.environ.pop(config.BUGWARRIORRC, None)
#         os.environ.pop('TASKRC', None)
#         os.environ.pop('XDG_CONFIG_HOME', None)
#         os.environ.pop('XDG_CONFIG_DIRS', None)
# 
#     def tearDown(self):
#         shutil.rmtree(self.tempdir, ignore_errors=True)
#         os.environ = self.old_environ
# 
#     @pytest.fixture(autouse=True)
#     def inject_fixtures(self, caplog):
#         self.caplog = caplog
# 
#     def validate(self):
#         conf = schema.validate_config(self.config, 'general', 'configpath')
#         conf['general'].data = data.BugwarriorData(self.lists_path)
#         conf['general'].interactive = False
#         return conf
# 
#     def assertValidationError(self, expected):
# 
#         with self.assertRaises(SystemExit):
#             self.validate()
# 
#         # Only one message should be logged.
#         self.assertEqual(len(self.caplog.records), 1)
# 
#         self.assertIn(expected, self.caplog.records[0].message)
# 
# Path: tests/test_github.py
# ARBITRARY_ISSUE = {
#     'title': 'Hallo',
#     'html_url': 'https://github.com/arbitrary_username/arbitrary_repo/pull/1',
#     'url': 'https://api.github.com/repos/arbitrary_username/arbitrary_repo/issues/1',
#     'number': 10,
#     'body': 'Something',
#     'user': {'login': 'arbitrary_login'},
#     'milestone': {'title': 'alpha'},
#     'labels': [{'name': 'bugfix'}],
#     'created_at': ARBITRARY_CREATED.isoformat(),
#     'closed_at': ARBITRARY_CLOSED.isoformat(),
#     'updated_at': ARBITRARY_UPDATED.isoformat(),
#     'repo': 'arbitrary_username/arbitrary_repo',
#     'state': 'closed'
# }
# 
# ARBITRARY_EXTRA = {
#     'project': 'one',
#     'type': 'issue',
#     'annotations': [],
#     'body': 'Something',
#     'namespace': 'arbitrary_username',
# }


import os
import logging
from unittest import mock
from click.testing import CliRunner
from bugwarrior import command
from bugwarrior.config.load import BugwarriorConfigParser
from .base import ConfigTest
from .test_github import ARBITRARY_ISSUE, ARBITRARY_EXTRA
            self.runner.invoke(command.pull)

        logs = [rec.message for rec in self.caplog.records]
        self.assertIn(
            'Aborted my_broken_service due to critical error.', logs)
        self.assertIn('Adding 0 tasks', logs)

    @mock.patch(
        'bugwarrior.services.github.GithubService.issues', fake_github_issues)
    @mock.patch('bugzilla.Bugzilla')
    def test_partial_failure_database_integrity(self, bugzillalib):
        """
        When a service fails and is terminated, don't close existing tasks.

        See https://github.com/ralphbean/bugwarrior/issues/821.
        """
        # Add the broken service to the configuration.
        self.config.set('general', 'targets', 'my_service,my_broken_service')
        self.config.add_section('my_broken_service')
        self.config.set('my_broken_service', 'service', 'bugzilla')
        self.config.set(
            'my_broken_service', 'bugzilla.base_uri', 'bugzilla.redhat.com')
        self.config.set(
            'my_broken_service', 'bugzilla.username', 'rbean@redhat.com')
        self.write_rc(self.config)

        # Add a task to each service.
        with self.caplog.at_level(logging.DEBUG):
            with mock.patch('bugwarrior.services.bz.BugzillaService.issues',
                            fake_bz_issues):

# Path: bugwarrior/command.py
# def _get_section_name(flavor):
# def _try_load_config(main_section, interactive=False):
# def pull(dry_run, flavor, interactive, debug):
# def vault():
# def targets():
# def list():
# def clear(target, username):
# def set(target, username):
# def uda(flavor):
# 
# Path: bugwarrior/config/load.py
# class BugwarriorConfigParser(configparser.ConfigParser):
#     def __init__(self, *args, allow_no_value=True, **kwargs):
#         super().__init__(*args, allow_no_value=allow_no_value, **kwargs)
# 
#     def getint(self, section, option):
#         """ Accepts both integers and empty values. """
#         try:
#             return super().getint(section, option)
#         except ValueError:
#             if self.get(section, option) == '':
#                 return None
#             else:
#                 raise ValueError(
#                     "{section}.{option} must be an integer or empty.".format(
#                         section=section, option=option))
# 
#     @staticmethod
#     def optionxform(option):
#         """ Do not lowercase key names. """
#         return option
# 
# Path: tests/base.py
# class ConfigTest(unittest.TestCase):
#     """
#     Creates config files, configures the environment, and cleans up afterwards.
#     """
#     def setUp(self):
#         self.old_environ = os.environ.copy()
#         self.tempdir = tempfile.mkdtemp(prefix='bugwarrior')
# 
#         # Create temporary config files.
#         self.taskrc = os.path.join(self.tempdir, '.taskrc')
#         self.lists_path = os.path.join(self.tempdir, 'lists')
#         os.mkdir(self.lists_path)
#         with open(self.taskrc, 'w+') as fout:
#             fout.write('data.location=%s\n' % self.lists_path)
# 
#         # Configure environment.
#         os.environ['HOME'] = self.tempdir
#         os.environ.pop(config.BUGWARRIORRC, None)
#         os.environ.pop('TASKRC', None)
#         os.environ.pop('XDG_CONFIG_HOME', None)
#         os.environ.pop('XDG_CONFIG_DIRS', None)
# 
#     def tearDown(self):
#         shutil.rmtree(self.tempdir, ignore_errors=True)
#         os.environ = self.old_environ
# 
#     @pytest.fixture(autouse=True)
#     def inject_fixtures(self, caplog):
#         self.caplog = caplog
# 
#     def validate(self):
#         conf = schema.validate_config(self.config, 'general', 'configpath')
#         conf['general'].data = data.BugwarriorData(self.lists_path)
#         conf['general'].interactive = False
#         return conf
# 
#     def assertValidationError(self, expected):
# 
#         with self.assertRaises(SystemExit):
#             self.validate()
# 
#         # Only one message should be logged.
#         self.assertEqual(len(self.caplog.records), 1)
# 
#         self.assertIn(expected, self.caplog.records[0].message)
# 
# Path: tests/test_github.py
# ARBITRARY_ISSUE = {
#     'title': 'Hallo',
#     'html_url': 'https://github.com/arbitrary_username/arbitrary_repo/pull/1',
#     'url': 'https://api.github.com/repos/arbitrary_username/arbitrary_repo/issues/1',
#     'number': 10,
#     'body': 'Something',
#     'user': {'login': 'arbitrary_login'},
#     'milestone': {'title': 'alpha'},
#     'labels': [{'name': 'bugfix'}],
#     'created_at': ARBITRARY_CREATED.isoformat(),
#     'closed_at': ARBITRARY_CLOSED.isoformat(),
#     'updated_at': ARBITRARY_UPDATED.isoformat(),
#     'repo': 'arbitrary_username/arbitrary_repo',
#     'state': 'closed'
# }
# 
# ARBITRARY_EXTRA = {
#     'project': 'one',
#     'type': 'issue',
#     'annotations': [],
#     'body': 'Something',
#     'namespace': 'arbitrary_username',
# }


# Path: tests/test_command.py
import os
import logging
from unittest import mock
from click.testing import CliRunner
from bugwarrior import command
from bugwarrior.config.load import BugwarriorConfigParser
from .base import ConfigTest
from .test_github import ARBITRARY_ISSUE, ARBITRARY_EXTRA

            self.runner.invoke(command.pull)

        logs = [rec.message for rec in self.caplog.records]
        self.assertIn(
            'Aborted my_broken_service due to critical error.', logs)
        self.assertIn('Adding 0 tasks', logs)

    @mock.patch(
        'bugwarrior.services.github.GithubService.issues', fake_github_issues)
    @mock.patch('bugzilla.Bugzilla')
    def test_partial_failure_database_integrity(self, bugzillalib):
        """
        When a service fails and is terminated, don't close existing tasks.

        See https://github.com/ralphbean/bugwarrior/issues/821.
        """
        # Add the broken service to the configuration.
        self.config.set('general', 'targets', 'my_service,my_broken_service')
        self.config.add_section('my_broken_service')
        self.config.set('my_broken_service', 'service', 'bugzilla')
        self.config.set(
            'my_broken_service', 'bugzilla.base_uri', 'bugzilla.redhat.com')
        self.config.set(
            'my_broken_service', 'bugzilla.username', 'rbean@redhat.com')
        self.write_rc(self.config)

        # Add a task to each service.
        with self.caplog.at_level(logging.DEBUG):
            with mock.patch('bugwarrior.services.bz.BugzillaService.issues',
                            fake_bz_issues):

                self.runner.invoke(command.pull)