Full migration to idiomatic pytest

bugwarrior/config/data.py

@@ -16,11 +16,12 @@ def get_data_path(taskrc: str | Path) -> str:
     env = dict(os.environ)
     env['TASKRC'] = str(taskrc)
 
-    tw_show = subprocess.Popen(('task', '_show'), stdout=subprocess.PIPE, env=env)
-    data_location = subprocess.check_output(
-        ('grep', '-e', '^' + line_prefix), stdin=tw_show.stdout
-    )
-    tw_show.wait()
+    with subprocess.Popen(
+        ('task', '_show'), stdout=subprocess.PIPE, env=env
+    ) as tw_show:
+        data_location = subprocess.check_output(
+            ('grep', '-e', '^' + line_prefix), stdin=tw_show.stdout
+        )
     data_path = data_location[len(line_prefix) :].rstrip().decode('utf-8')
 
     if not data_path:

bugwarrior/config/ini2toml_plugin.py

@@ -1,3 +1,4 @@
+import functools
 import importlib
 import inspect
 import logging
@@ -53,10 +54,18 @@ def get_field_type(attrs: dict) -> typing.Optional[str]:
     return None
 
 
+@functools.cache
+def _schema_properties(schema: type[pydantic.BaseModel]) -> dict:
+    # model_json_schema() rebuilds the schema from scratch on every call, and
+    # this is invoked once per config example in the docs build (~150+
+    # times), so cache it per schema class.
+    return schema.model_json_schema()['properties']
+
+
 def convert_section(
     section: IntermediateRepr, schema: type[pydantic.BaseModel]
 ) -> None:
-    for prop, attrs in schema.model_json_schema()['properties'].items():
+    for prop, attrs in _schema_properties(schema).items():
         field_type = get_field_type(attrs)
         if field_type == 'boolean':
             to_bool(section, prop)
@@ -66,6 +75,20 @@ def convert_section(
             to_list(section, prop)
 
 
+@functools.cache
+def _service_schema(service: str) -> type[ServiceConfig]:
+    # Resolving a service's config class scans the module with
+    # inspect.getmembers() on every call, and this is invoked once per
+    # config example in the docs build (~150+ times), so cache it per
+    # service name.
+    module_name = {'bugzilla': 'bz', 'phabricator': 'phab'}.get(service, service)
+    service_module = importlib.import_module(f'bugwarrior.services.{module_name}')
+    for _, obj in inspect.getmembers(service_module, predicate=inspect.isclass):
+        if issubclass(obj, ServiceConfig):
+            return obj
+    raise ValueError(f"ServiceConfig class not found in {service} module.")
+
+
 def process_values(doc: IntermediateRepr) -> IntermediateRepr:
     for name, section in doc.items():
         if isinstance(name, str):
@@ -96,22 +119,7 @@ def process_values(doc: IntermediateRepr) -> IntermediateRepr:
                         section.rename(key, newkey)
 
                 # Get Config
-                module_name = {'bugzilla': 'bz', 'phabricator': 'phab'}.get(
-                    service, service
-                )
-                service_module = importlib.import_module(
-                    f'bugwarrior.services.{module_name}'
-                )
-                for name, obj in inspect.getmembers(
-                    service_module, predicate=inspect.isclass
-                ):
-                    if issubclass(obj, ServiceConfig):
-                        schema = obj
-                        break
-                else:
-                    raise ValueError(
-                        f"ServiceConfig class not found in {service} module."
-                    )
+                schema = _service_schema(service)
 
                 # Convert Types
                 convert_section(section, schema)

bugwarrior/config/load.py

@@ -50,15 +50,19 @@ def get_config_path() -> str:
 
 
 def format_config(config: dict) -> dict[str, Any]:
+    config = config.copy()
+    formatted: dict[str, Any] = {}
+    if "flavor" in config:
+        formatted["flavor"] = {**config.pop("flavor")}
     if "general" in config:
-        config.setdefault("flavor", {})["general"] = config.pop("general")
-
-    config["services"] = [
-        {**config.pop(section), "target": section}
-        for section in list(config)
-        if section not in {"hooks", "notifications", "flavor"}
+        formatted.setdefault("flavor", {})["general"] = config.pop("general")
+    for key in ("hooks", "notifications"):
+        if key in config:
+            formatted[key] = config.pop(key)
+    formatted["services"] = [
+        {**config.pop(section), "target": section} for section in list(config)
     ]
-    return config
+    return formatted
 
 
 def parse_toml_file(configpath: str) -> dict[str, Any]:

bugwarrior/config/secrets.py

@@ -90,15 +90,13 @@ def oracle_eval(command: str) -> str:
     p = subprocess.Popen(
         command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
     )
-    p.wait()
-    assert p.stdout is not None
-    assert p.stderr is not None
+    stdout, stderr = p.communicate()
     if p.returncode == 0:
-        return p.stdout.readline().strip().decode('utf-8')
+        return stdout.split(b'\n', 1)[0].strip().decode('utf-8')
     else:
         log.critical(
             "Error retrieving password: `{command}` returned '{error}'".format(
-                command=command, error=p.stderr.read().strip()
+                command=command, error=stderr.strip()
             )
         )
         sys.exit(1)

bugwarrior/docs/_ext/config.py

@@ -1,9 +1,20 @@
+import functools
+
 from ini2toml.api import Translator
 from sphinx.directives.code import CodeBlock
 from sphinx.util.docutils import SphinxDirective
 from sphinx_inline_tabs._impl import TabDirective
 
 
+@functools.cache
+def _translator() -> Translator:
+    # Translator() discovers its ini/toml plugins via importlib.metadata
+    # entry points on every instantiation, which is expensive. The plugin
+    # set is fixed for the lifetime of the process, so build one and reuse
+    # it across every ``.. config::`` directive instead of once per call.
+    return Translator()
+
+
 class Config(SphinxDirective):
     optional_arguments = 1
     option_spec = {'fragment': str}  # section schema to stub
@@ -29,7 +40,7 @@ def run(self):
                 '[some_section]\nservice = ' + self.options['fragment'] + '\n'
             )
             initext = stub_section + initext
-        tomltext = Translator().translate(initext, 'bugwarriorrc')
+        tomltext = _translator().translate(initext, 'bugwarriorrc')
         if 'fragment' in self.options:  # remove stub
             stub_len = len(stub_section) + 2  # toml adds quotes to strings
             tomltext = tomltext[stub_len:]

bugwarrior/docs/other-services/tutorial.rst

@@ -247,54 +247,71 @@ If you're developing your service in a separate package, it's time to create a `
 If you're developing in the bugwarrior repo, you can simply add your entry to the existing ``[project.entry-points."bugwarrior.service"]`` table.
 
 8. Tests
-----------
+--------
 
 .. note::
 
    The remainder of this tutorial is not geared towards third-party services. While you are free to use bugwarrior's testing infrastructure, no attempt is being made to maintain the stability of these interfaces at this time.
 
-
-
-Create a test file and implement at least the minimal service tests by inheriting from ``ServiceIssueTest``.
+Create a test file. Declare ``SERVICE_CLASS`` and ``SERVICE_CONFIG`` at module level -- these are picked up by the ``service`` and ``make_service`` fixtures shared across all service tests (see ``tests/services/conftest.py``), which build a mock service instance for you. Fake record data belongs in a ``record`` fixture rather than instance state, since a fresh dictionary per test avoids accidental sharing between tests.
 
 .. code:: bash
 
    touch tests/services/test_gitbug.py
 
 .. code:: python
 
-  class TestGitBugIssue(ServiceIssueTest):
-      SERVICE_CONFIG = {
-          'service': 'gitbug',
-          'path': '/dev/null',
-      }
+  from unittest import mock
+
+  import pytest
 
-      def setUp(self):
-          super().setUp()
+  from bugwarrior.collect import TaskConstructor
+  from bugwarrior.services.gitbug import GitBugClient, GitBugService
 
-          self.data = TestData()
+  SERVICE_CLASS = GitBugService
 
-          self.service = self.get_mock_service(GitBugService)
-          self.service.client = mock.MagicMock(spec=GitBugClient)
-          self.service.client.get_issues = mock.MagicMock(
-              return_value=[self.data.arbitrary_bug])
+  SERVICE_CONFIG = {
+      'service': 'gitbug',
+      'path': '/dev/null',
+  }
 
-      def test_to_taskwarrior(self):
-          issue = self.service.get_issue_for_record(
-              self.data.arbitrary_bug, {})
+
+  @pytest.fixture
+  def record():
+      return {
+          'id': 'arbitrary_id',
+          'title': 'arbitrary_title',
+          'state': 'open',
+          'author': {'name': 'arbitrary_author'},
+          'labels': [],
+          'createdAt': '2016-06-06T06:07:08.123-0700',
+      }
+
+
+  class TestGitBugIssue:
+      @pytest.fixture
+      def service(self, make_service):
+          service = make_service()
+          service.client = mock.MagicMock(spec=GitBugClient)
+          return service
+
+      def test_to_taskwarrior(self, service, record):
+          issue = service.get_issue_for_record(record, {})
 
           expected = { ... }
 
           actual = issue.to_taskwarrior()
 
-          self.assertEqual(actual, expected)
+          assert actual == expected
+
+      def test_issues(self, service, record):
+          service.client.get_issues.return_value = [record]
 
-      def test_issues(self):
-          issue = next(self.service.issues())
+          issue = next(service.issues())
 
           expected = { ... }
 
-          self.assertEqual(TaskConstructor(issue).get_taskwarrior_record(), expected)
+          assert TaskConstructor(issue).get_taskwarrior_record() == expected
 
 9. Documentation
 ------------------

pyproject.toml

@@ -88,17 +88,23 @@ bugwarrior = "bugwarrior.config.ini2toml_plugin:activate"
 
 [tool.uv]
 exclude-newer = "3 days"
+# tomlkit 0.15.1's parser rewrite changed comment-whitespace serialization
+# (undocumented in its changelog): blank comment lines lose a trailing
+# space. tests/config/example-bugwarrior.toml reflects this new output;
+# pin forward so we never silently regress to the old 0.15.0 formatting.
+constraint-dependencies = ["tomlkit>=0.15.1"]
 
 [tool.ruff.format]
 quote-style = "preserve"
 skip-magic-trailing-comma = true
 
 [tool.ruff.lint]
-select = ["E4", "E7", "E9", "F", "W", "I", "E501", "ANN"]
+select = ["E4", "E7", "E9", "F", "W", "I", "E501", "ANN", "PT"]
 ignore = ["ANN401"]
 # ANN rules (flake8-annotations) detect untyped code — useful for gradual typing.
 # Enforced in application code under bugwarrior/ (excluding tests/ and bugwarrior/docs/).
 # ANN401 (disallow Any) is too strict for wrappers, *args/**kwargs, and data storage.
+# PT rules (flake8-pytest-style) enforce pytest conventions in the test suite.
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**" = ["ANN"]
@@ -115,6 +121,15 @@ split-on-trailing-comma = false
 [tool.ruff.lint.pycodestyle]
 max-line-length = 100
 
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+filterwarnings = [
+    "error",
+    # taskw still uses deprecated distutils version classes; show the warning
+    # without failing the suite.
+    "default::DeprecationWarning:taskw",
+]
+
 [tool.setuptools.packages.find]
 include = ["bugwarrior*"]
 

tests/base.py

@@ -1,17 +1,14 @@
 import contextlib
-import os.path
-import shutil
-import tempfile
 import typing
-import unittest
 import unittest.mock
 
-import pytest
-
 from bugwarrior import config, services
+from bugwarrior.collect import get_service_instances
 from bugwarrior.config import validation
 from bugwarrior.config.load import format_config
 
+from .services.base import get_mock_service
+
 
 class DumbConfig(config.ServiceConfig):
     service: typing.Literal["test"] = "test"
@@ -59,6 +56,13 @@ def issues(self):
         raise NotImplementedError
 
 
+def make_issue(general_overrides=None, config_overrides=None):
+    service = get_mock_service(
+        DumbService, config_overrides, general_overrides=general_overrides
+    )
+    return service.get_issue_for_record({})
+
+
 #: Modules that import get_service by name and must be patched together so the
 #: fake service resolves consistently across config loading, collection, and db.
 _GET_SERVICE_MODULES = (
@@ -99,53 +103,10 @@ def fake_get_service(name):
         yield
 
 
-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['XDG_CONFIG_HOME'] = os.path.join(self.tempdir, '.config')
-        os.environ.pop(config.BUGWARRIORRC, None)
-        os.environ.pop('TASKRC', None)
-        os.environ.pop('XDG_CONFIG_DIRS', None)
-
-    def tearDown(self):
-        shutil.rmtree(self.tempdir, ignore_errors=True)
-
-        os.environ.clear()
-        os.environ.update(self.old_environ)
-
-    @pytest.fixture(autouse=True)
-    def inject_fixtures(self, caplog):
-        self.caplog = caplog
-
-    def validate(self) -> validation.Config:
-        config = self.config.copy()
-        config['general'] = config.get('general', {})
-        formatted_config = format_config(config)
-        return validation.validate_config(formatted_config, 'general', 'configpath')
-
-    def assertValidationError(self, expected):
-        with pytest.raises(SystemExit):
-            self.validate()
-
-        # Only one message should be logged.
-        assert len(self.caplog.records) == 1
+def validate(config) -> validation.Config:
+    formatted_config = format_config(config)
+    return validation.validate_config(formatted_config, 'general', 'configpath')
 
-        assert expected in self.caplog.records[0].message
 
-        # We may want to use this assertion more than once per test.
-        self.caplog.clear()
+def get_validated_service(config):
+    return get_service_instances(validate(config))[0]