fix: several test improvements

Author: Lotram

tests/base.py

@@ -1,6 +1,7 @@
 import contextlib
 import os.path
 import shutil
+import sys
 import tempfile
 import typing
 import unittest
@@ -96,7 +97,7 @@ def fake_get_service(name):
             stack.enter_context(
                 unittest.mock.patch(f'{module}.get_service', fake_get_service)
             )
-        yield mapping
+        yield
 
 
 class ConfigTest(unittest.TestCase):
@@ -132,15 +133,13 @@ def tearDown(self):
     def inject_fixtures(self, caplog):
         self.caplog = caplog
 
-    def enter_context(self, cm):
-        """Enter a context manager for the duration of the test.
+    if sys.version_info < (3, 11):
 
-        Backport of unittest.TestCase.enterContext, which is only available on
-        Python 3.11+ (we still support 3.10).
-        """
-        value = cm.__enter__()
-        self.addCleanup(cm.__exit__, None, None, None)
-        return value
+        def enterContext(self, cm):
+            """Backport of unittest.TestCase.enterContext for Python 3.10."""
+            value = cm.__enter__()
+            self.addCleanup(cm.__exit__, None, None, None)
+            return value
 
     def validate(self) -> validation.Config:
         config = self.config.copy()

tests/config/test_validation.py

@@ -42,7 +42,7 @@ class ValidationService(DumbService):
 class TestValidation(ConfigTest):
     def setUp(self):
         super().setUp()
-        self.enter_context(register_services({"test": ValidationService}))
+        self.enterContext(register_services({"test": ValidationService}))
         self.config = {
             "general": {"targets": ["my_service"]},
             "my_service": {"service": "test", "username": "ralph"},
@@ -83,6 +83,16 @@ def test_service_missing(self):
 
         self.assertValidationError("No option 'service' in section: 'my_service'")
 
+    def test_service_missing_without_other_service(self):
+        del self.config['my_service']['service']
+
+        self.assertValidationError(
+            "2 validation errors found in configpath\n"
+            "See https://bugwarrior.readthedocs.io\n\n"
+            "[my_service]  <- Field required\n"
+            "[my_service]  <- unrecognized option"
+        )
+
     def test_extra_field(self):
         """Undeclared fields are forbidden."""
         self.config['my_service']['undeclared_field'] = 'extra'
@@ -135,8 +145,6 @@ def test_deprecated_filter_merge_requests_and_include_merge_requests(self):
 
     def test_deprecated_project_name(self):
         """We're just testing that deprecation doesn't break validation."""
-        self.validate()
-
         self.config['my_service']['project_name'] = 'myproject'
         self.validate()
 

tests/test_command.py

@@ -41,17 +41,22 @@ class SecondaryService(DumbService):
     ISSUE_CLASS = SecondaryIssue
 
 
-ARBITRARY_RECORD = {
-    'title': 'Hallo',
-    'url': 'https://example.com',
-    'number': 10,
-    'labels': [],
-}
-ARBITRARY_EXTRA = {'project': 'one', 'type': 'issue', 'annotations': []}
+def yields_one(target_specific_url=False):
+    def issues(self):
+        record = {
+            'title': 'Hallo',
+            'url': 'https://example.com',
+            'number': 10,
+            'labels': [],
+        }
+
+        if target_specific_url:
+            record['url'] = f'https://example.com/{self.config.target}'
 
+        extra = {'project': 'one', 'type': 'issue', 'annotations': []}
+        yield self.get_issue_for_record(record, extra)
 
-def yields_one(self):
-    yield self.get_issue_for_record(ARBITRARY_RECORD, ARBITRARY_EXTRA)
+    return issues
 
 
 def yields_none(self):
@@ -62,12 +67,6 @@ def raises(self):
     raise Exception('message')
 
 
-def yields_per_target(self):
-    """Yield one issue whose url is unique per target."""
-    record = dict(ARBITRARY_RECORD, url=f'https://example.com/{self.config.target}')
-    yield self.get_issue_for_record(record, ARBITRARY_EXTRA)
-
-
 def fake_service(issues, base=DumbService):
     """
     Build a fake service class whose issues() is the given function.
@@ -106,9 +105,10 @@ def test_success(self):
         """
         A normal `bugwarrior pull` invocation.
         """
-        self.enter_context(register_services({'test': fake_service(yields_one)}))
-
-        with self.caplog.at_level(logging.INFO):
+        with (
+            register_services({'test': fake_service(yields_one())}),
+            self.caplog.at_level(logging.INFO),
+        ):
             self.runner.invoke(command.cli, args=('pull', '--debug'))
 
         logs = [rec.message for rec in self.caplog.records]
@@ -121,9 +121,10 @@ def test_failure(self):
         """
         A broken `bugwarrior pull` invocation.
         """
-        self.enter_context(register_services({'test': fake_service(raises)}))
-
-        with self.caplog.at_level(logging.ERROR):
+        with (
+            register_services({'test': fake_service(raises)}),
+            self.caplog.at_level(logging.ERROR),
+        ):
             self.runner.invoke(command.cli, args=('pull', '--debug'))
 
         self.assertNotEqual(self.caplog.records, [])
@@ -147,16 +148,15 @@ def test_partial_failure_survival(self):
         self.config['my_broken_service'] = {'service': 'secondary'}
         self.write_rc(self.config)
 
-        self.enter_context(
+        with (
             register_services(
                 {
                     'test': fake_service(yields_none),
                     'secondary': fake_service(raises, base=SecondaryService),
                 }
-            )
-        )
-
-        with self.caplog.at_level(logging.INFO):
+            ),
+            self.caplog.at_level(logging.INFO),
+        ):
             self.runner.invoke(command.cli, args=('pull', '--debug'))
 
         logs = [rec.message for rec in self.caplog.records]
@@ -175,23 +175,23 @@ def test_partial_failure_database_integrity(self):
 
         # Add a task to each service.
         both_working = {
-            'test': fake_service(yields_per_target),
-            'secondary': fake_service(yields_per_target, base=SecondaryService),
+            'test': fake_service(yields_one(target_specific_url=True)),
+            'secondary': fake_service(
+                yields_one(target_specific_url=True), base=SecondaryService
+            ),
         }
-        with register_services(both_working):
-            with self.caplog.at_level(logging.DEBUG):
-                self.runner.invoke(command.cli, args=('pull', '--debug'))
+        with register_services(both_working), self.caplog.at_level(logging.DEBUG):
+            self.runner.invoke(command.cli, args=('pull', '--debug'))
         logs = [rec.message for rec in self.caplog.records]
         self.assertIn('Adding 2 tasks', logs)
 
         # Break the secondary service and run pull again.
         secondary_broken = {
-            'test': fake_service(yields_per_target),
+            'test': fake_service(yields_one(target_specific_url=True)),
             'secondary': fake_service(raises, base=SecondaryService),
         }
-        with register_services(secondary_broken):
-            with self.caplog.at_level(logging.INFO):
-                self.runner.invoke(command.cli, args=('pull', '--debug'))
+        with register_services(secondary_broken), self.caplog.at_level(logging.INFO):
+            self.runner.invoke(command.cli, args=('pull', '--debug'))
         logs = [rec.message for rec in self.caplog.records]
 
         # Make sure my_broken_service failed while my_service succeeded.
@@ -209,13 +209,15 @@ def test_locked_repository(self, file_lock):
         """
         # The service is never collected (the lock fails first), but config
         # loading still resolves it.
-        self.enter_context(register_services({'test': DumbService}))
         lockfile_path = pathlib.Path(self.lists_path) / 'bugwarrior.lockfile'
         file_lock.return_value.__enter__.side_effect = command.Timeout(
             str(lockfile_path)
         )
 
-        with self.caplog.at_level(logging.CRITICAL):
+        with (
+            register_services({'test': DumbService}),
+            self.caplog.at_level(logging.CRITICAL),
+        ):
             result = self.runner.invoke(command.cli, args=('pull', '--debug'))
 
         self.assertEqual(result.exit_code, 1)
@@ -231,9 +233,10 @@ def test_legacy_cli(self):
 
         Also test that it logs a deprecation warning.
         """
-        self.enter_context(register_services({'test': fake_service(yields_one)}))
-
-        with self.caplog.at_level(logging.INFO):
+        with (
+            register_services({'test': fake_service(yields_one())}),
+            self.caplog.at_level(logging.INFO),
+        ):
             self.runner.invoke(command.pull, args=('--debug'))
 
         logs = [rec.message for rec in self.caplog.records]

tests/test_db.py

@@ -63,8 +63,7 @@ def test_handles_missing_tags(self):
 class TestSynchronize(ConfigTest):
     def setUp(self):
         super().setUp()
-        # synchronize() resolves the service via get_service for UNIQUE_KEY/UDAS.
-        self.enter_context(register_services())
+        self.enterContext(register_services())
         self.bwconfig = Config(
             service_configs=[DumbConfig(target='my_service')],
             main=schema.MainSectionConfig(