[gettext] fix flaky test on windows (#11940)

tests/test_intl/test_intl.py

@@ -8,7 +8,6 @@ import os.path
 import re
 import shutil
 import time
-from pathlib import Path
 
 import pytest
 from babel.messages import mofile, pofile
@@ -19,10 +18,12 @@ from sphinx import locale
 from sphinx.testing.util import assert_node, etree_parse, strip_escseq
 from sphinx.util.nodes import NodeMatcher
 
+_CATALOG_LOCALE = 'xx'
+
 sphinx_intl = pytest.mark.sphinx(
     testroot='intl',
     confoverrides={
-        'language': 'xx', 'locale_dirs': ['.'],
+        'language': _CATALOG_LOCALE, 'locale_dirs': ['.'],
         'gettext_compact': False,
     },
 )
@@ -38,22 +39,20 @@ def write_mo(pathname, po):
         return mofile.write_mo(f, po)
 
 
-@pytest.fixture(autouse=True)
-def _setup_intl(app_params):
-    assert isinstance(app_params.kwargs['srcdir'], Path)
-    srcdir = app_params.kwargs['srcdir']
-    for dirpath, _dirs, files in os.walk(srcdir):
-        dirpath = Path(dirpath)
-        for f in [f for f in files if f.endswith('.po')]:
-            po = str(dirpath / f)
-            mo = srcdir / 'xx' / 'LC_MESSAGES' / (
-                os.path.relpath(po[:-3], srcdir) + '.mo')
-            if not mo.parent.exists():
-                mo.parent.mkdir(parents=True, exist_ok=True)
+def _set_mtime_ns(target, value):
+    os.utime(target, ns=(value, value))
+    return os.stat(target).st_mtime_ns
 
-            if not mo.exists() or os.stat(mo).st_mtime < os.stat(po).st_mtime:
-                # compile .mo file only if needed
-                write_mo(mo, read_po(po))
+
+def _get_bom_intl_path(app):
+    basedir = app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES'
+    return basedir / 'bom.po', basedir / 'bom.mo'
+
+
+def _get_update_targets(app):
+    app.env.find_files(app.config, app.builder)
+    added, changed, removed = app.env.get_outdated_files(config_changed=False)
+    return added, changed, removed
 
 
 @pytest.fixture(autouse=True)
@@ -316,7 +315,7 @@ def test_text_glossary_term_inconsistencies(app, warning):
 def test_gettext_section(app):
     app.build()
     # --- section
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'section.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'section.po')
     actual = read_po(app.outdir / 'section.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -329,7 +328,7 @@ def test_text_section(app):
     app.build()
     # --- section
     result = (app.outdir / 'section.txt').read_text(encoding='utf8')
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'section.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'section.po')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.string in result
 
@@ -468,12 +467,12 @@ def test_text_admonitions(app):
 def test_gettext_toctree(app):
     app.build()
     # --- toctree (index.rst)
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'index.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'index.po')
     actual = read_po(app.outdir / 'index.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
     # --- toctree (toctree.rst)
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'toctree.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'toctree.po')
     actual = read_po(app.outdir / 'toctree.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -485,7 +484,7 @@ def test_gettext_toctree(app):
 def test_gettext_table(app):
     app.build()
     # --- toctree
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'table.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'table.po')
     actual = read_po(app.outdir / 'table.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -498,7 +497,7 @@ def test_text_table(app):
     app.build()
     # --- toctree
     result = (app.outdir / 'table.txt').read_text(encoding='utf8')
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'table.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'table.po')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.string in result
 
@@ -515,7 +514,7 @@ def test_text_toctree(app):
     assert 'TABLE OF CONTENTS' in result
     # --- toctree (toctree.rst)
     result = (app.outdir / 'toctree.txt').read_text(encoding='utf8')
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'toctree.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'toctree.po')
     for expect_msg in (m for m in expect if m.id):
         assert expect_msg.string in result
 
@@ -526,7 +525,7 @@ def test_text_toctree(app):
 def test_gettext_topic(app):
     app.build()
     # --- topic
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'topic.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'topic.po')
     actual = read_po(app.outdir / 'topic.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -539,7 +538,7 @@ def test_text_topic(app):
     app.build()
     # --- topic
     result = (app.outdir / 'topic.txt').read_text(encoding='utf8')
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'topic.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'topic.po')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.string in result
 
@@ -550,7 +549,7 @@ def test_text_topic(app):
 def test_gettext_definition_terms(app):
     app.build()
     # --- definition terms: regression test for #2198, #2205
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'definition_terms.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'definition_terms.po')
     actual = read_po(app.outdir / 'definition_terms.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -562,7 +561,7 @@ def test_gettext_definition_terms(app):
 def test_gettext_glossary_terms(app, warning):
     app.build()
     # --- glossary terms: regression test for #1090
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'glossary_terms.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'glossary_terms.po')
     actual = read_po(app.outdir / 'glossary_terms.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -576,7 +575,7 @@ def test_gettext_glossary_terms(app, warning):
 def test_gettext_glossary_term_inconsistencies(app):
     app.build()
     # --- glossary term inconsistencies: regression test for #1090
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'glossary_terms_inconsistency.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'glossary_terms_inconsistency.po')
     actual = read_po(app.outdir / 'glossary_terms_inconsistency.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -588,7 +587,7 @@ def test_gettext_glossary_term_inconsistencies(app):
 def test_gettext_literalblock(app):
     app.build()
     # --- gettext builder always ignores ``only`` directive
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'literalblock.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'literalblock.po')
     actual = read_po(app.outdir / 'literalblock.pot')
     for expect_msg in [m for m in expect if m.id]:
         if len(expect_msg.id.splitlines()) == 1:
@@ -604,7 +603,7 @@ def test_gettext_literalblock(app):
 def test_gettext_buildr_ignores_only_directive(app):
     app.build()
     # --- gettext builder always ignores ``only`` directive
-    expect = read_po(app.srcdir / 'xx' / 'LC_MESSAGES' / 'only.po')
+    expect = read_po(app.srcdir / _CATALOG_LOCALE / 'LC_MESSAGES' / 'only.po')
     actual = read_po(app.outdir / 'only.pot')
     for expect_msg in [m for m in expect if m.id]:
         assert expect_msg.id in [m.id for m in actual if m.id]
@@ -633,7 +632,7 @@ def test_translation_progress_substitution(app):
 
 
 @pytest.mark.sphinx(testroot='intl', freshenv=True, confoverrides={
-    'language': 'xx', 'locale_dirs': ['.'],
+    'language': _CATALOG_LOCALE, 'locale_dirs': ['.'],
     'gettext_compact': False,
     'translation_progress_classes': True,
 })
@@ -681,50 +680,188 @@ def test_translation_progress_classes_true(app):
     assert len(doctree[0]) == 20
 
 
+class _MockClock:
+    """Object for mocking :func:`time.time_ns` (if needed).
+
+    Use :meth:`sleep` to make this specific clock sleep for some time.
+    """
+
+    def time(self) -> int:
+        """Nanosecond since 'fake' epoch."""
+        raise NotImplementedError
+
+    def sleep(self, ds: float) -> None:
+        """Sleep *ds* seconds."""
+        raise NotImplementedError
+
+
+class _MockWindowsClock(_MockClock):
+    """Object for mocking :func:`time.time_ns` on Windows platforms.
+
+    The result is in 'nanoseconds' but with a microsecond resolution
+    so that the division by 1_000 does not cause rounding issues.
+    """
+
+    def __init__(self) -> None:
+        self.us: int = 0  # current microsecond 'tick'
+
+    def time(self) -> int:
+        ret = 1_000 * self.us
+        self.us += 1
+        return ret
+
+    def sleep(self, ds: float) -> None:
+        self.us += int(ds * 1e6)
+
+
+class _MockUnixClock(_MockClock):
+    """Object for mocking :func:`time.time_ns` on Unix platforms.
+
+    Since nothing is needed for Unix platforms, this object acts as
+    a proxy so that the API is the same as :class:`_MockWindowsClock`.
+    """
+
+    def time(self) -> int:
+        return time.time_ns()
+
+    def sleep(self, ds: float) -> None:
+        time.sleep(ds)
+
+
+@pytest.fixture()
+def mock_time_and_i18n(
+    monkeypatch: pytest.MonkeyPatch,
+) -> tuple[pytest.MonkeyPatch, _MockClock]:
+    from sphinx.util.i18n import CatalogInfo
+
+    # save the 'original' definition
+    catalog_write_mo = CatalogInfo.write_mo
+
+    def mock_write_mo(self, locale, use_fuzzy=False):
+        catalog_write_mo(self, locale, use_fuzzy)
+        # ensure that the .mo file being written has a correct fake timestamp
+        _set_mtime_ns(self.mo_path, time.time_ns())
+
+    # see: https://github.com/pytest-dev/pytest/issues/363
+    with pytest.MonkeyPatch.context() as mock:
+        if os.name == 'posix':
+            clock = _MockUnixClock()
+        else:
+            # When using pytest.mark.parametrize() to emulate test repetition,
+            # the teardown phase on Windows fails due to an error apparently in
+            # the colorama.ansitowin32 module, so we forcibly disable colors.
+            mock.setenv('NO_COLOR', '1')
+            # apply the patch only for Windows
+            clock = _MockWindowsClock()
+            mock.setattr('time.time_ns', clock.time)
+            # Use clock.sleep() to emulate time.sleep() but do not try
+            # to mock the latter since this might break other libraries.
+            mock.setattr('sphinx.util.i18n.CatalogInfo.write_mo', mock_write_mo)
+        yield mock, clock
+
+
 @sphinx_intl
-# use individual shared_result directory to avoid "incompatible doctree" error
-@pytest.mark.sphinx(testroot='builder-gettext-dont-rebuild-mo')
-def test_gettext_dont_rebuild_mo(make_app, app_params):
-    # --- don't rebuild by .mo mtime
-    def get_update_targets(app_):
-        app_.env.find_files(app_.config, app_.builder)
-        added, changed, removed = app_.env.get_outdated_files(config_changed=False)
-        return added, changed, removed
+# use the same testroot as 'gettext' since the latter contains less PO files
+@pytest.mark.sphinx('dummy', testroot='builder-gettext-dont-rebuild-mo', freshenv=True)
+def test_dummy_should_rebuild_mo(mock_time_and_i18n, make_app, app_params):
+    mock, clock = mock_time_and_i18n
+    assert os.name == 'posix' or clock.time() == 0
 
     args, kwargs = app_params
+    app = make_app(*args, **kwargs)
+    po_path, mo_path = _get_bom_intl_path(app)
+
+    # creation time of the those files (order does not matter)
+    bom_rst = app.srcdir / 'bom.rst'
+    bom_rst_time = time.time_ns()
+
+    index_rst = app.srcdir / 'index.rst'
+    index_rst_time = time.time_ns()
+    po_time = time.time_ns()
+
+    # patch the 'creation time' of the source files
+    assert _set_mtime_ns(po_path, po_time) == po_time
+    assert _set_mtime_ns(bom_rst, bom_rst_time) == bom_rst_time
+    assert _set_mtime_ns(index_rst, index_rst_time) == index_rst_time
+
+    assert not mo_path.exists()
+    # when writing mo files, the counter is updated by calling
+    # patch_write_mo which is called to create .mo files (and
+    # thus the timestamp of the files are not those given by
+    # the OS but our fake ones)
+    app.build()
+    assert mo_path.exists()
+    # Do a real sleep on POSIX, or simulate a sleep on Windows
+    # to ensure that calls to time.time_ns() remain consistent.
+    clock.sleep(0.1 if os.name == 'posix' else 1)
+
+    # check that the source files were not modified
+    assert bom_rst.stat().st_mtime_ns == bom_rst_time
+    assert index_rst.stat().st_mtime_ns == index_rst_time
+    # check that the 'bom' document is discovered after the .mo
+    # file has been written on the disk (i.e., read_doc() is called
+    # after the creation of the .mo files)
+    assert app.env.all_docs['bom'] > mo_path.stat().st_mtime_ns // 1000
 
-    # phase1: build document with non-gettext builder and generate mo file in srcdir
-    app0 = make_app('dummy', *args, **kwargs)
-    app0.build()
-    time.sleep(0.01)
-    assert (app0.srcdir / 'xx' / 'LC_MESSAGES' / 'bom.mo').exists()
     # Since it is after the build, the number of documents to be updated is 0
-    update_targets = get_update_targets(app0)
-    assert update_targets[1] == set(), update_targets
+    update_targets = _get_update_targets(app)
+    assert update_targets[1] == set()
     # When rewriting the timestamp of mo file, the number of documents to be
     # updated will be changed.
-    mtime = (app0.srcdir / 'xx' / 'LC_MESSAGES' / 'bom.mo').stat().st_mtime
-    os.utime(app0.srcdir / 'xx' / 'LC_MESSAGES' / 'bom.mo', (mtime + 5, mtime + 5))
-    update_targets = get_update_targets(app0)
-    assert update_targets[1] == {'bom'}, update_targets
+    new_mo_time = time.time_ns()
+    assert _set_mtime_ns(mo_path, new_mo_time) == new_mo_time
+    update_targets = _get_update_targets(app)
+    assert update_targets[1] == {'bom'}
+    mock.undo()  # explicit call since it's not a context
 
-    # Because doctree for gettext builder can not be shared with other builders,
-    # erase doctreedir before gettext build.
-    shutil.rmtree(app0.doctreedir)
+    # remove all sources for the next test
+    shutil.rmtree(app.srcdir, ignore_errors=True)
+    time.sleep(0.1 if os.name == 'posix' else 0.5)  # real sleep
 
-    # phase2: build document with gettext builder.
+
+@sphinx_intl
+@pytest.mark.sphinx('gettext', testroot='builder-gettext-dont-rebuild-mo', freshenv=True)
+def test_gettext_dont_rebuild_mo(mock_time_and_i18n, app):
+    mock, clock = mock_time_and_i18n
+    assert os.name == 'posix' or clock.time() == 0
+
+    assert app.srcdir.exists()
+
+    # patch the 'creation time' of the source files
+    bom_rst = app.srcdir / 'bom.rst'
+    bom_rst_time = time.time_ns()
+    assert _set_mtime_ns(bom_rst, bom_rst_time) == bom_rst_time
+
+    index_rst = app.srcdir / 'index.rst'
+    index_rst_time = time.time_ns()
+    assert _set_mtime_ns(index_rst, index_rst_time) == index_rst_time
+
+    # phase 1: create fake MO file in the src directory
+    po_path, mo_path = _get_bom_intl_path(app)
+    write_mo(mo_path, read_po(po_path))
+    po_time = time.time_ns()
+    assert _set_mtime_ns(po_path, po_time) == po_time
+
+    # phase 2: build document with gettext builder.
     # The mo file in the srcdir directory is retained.
-    app = make_app('gettext', *args, **kwargs)
     app.build()
-    time.sleep(0.01)
+    # Do a real sleep on POSIX, or simulate a sleep on Windows
+    # to ensure that calls to time.time_ns() remain consistent.
+    clock.sleep(0.5 if os.name == 'posix' else 1)
     # Since it is after the build, the number of documents to be updated is 0
-    update_targets = get_update_targets(app)
-    assert update_targets[1] == set(), update_targets
+    update_targets = _get_update_targets(app)
+    assert update_targets[1] == set()
     # Even if the timestamp of the mo file is updated, the number of documents
     # to be updated is 0. gettext builder does not rebuild because of mo update.
-    os.utime(app0.srcdir / 'xx' / 'LC_MESSAGES' / 'bom.mo', (mtime + 10, mtime + 10))
-    update_targets = get_update_targets(app)
-    assert update_targets[1] == set(), update_targets
+    new_mo_time = time.time_ns()
+    assert _set_mtime_ns(mo_path, new_mo_time) == new_mo_time
+    update_targets = _get_update_targets(app)
+    assert update_targets[1] == set()
+    mock.undo()  # remove the patch
+
+    # remove all sources for the next test
+    shutil.rmtree(app.srcdir, ignore_errors=True)
+    time.sleep(0.1 if os.name == 'posix' else 0.5)  # real sleep
 
 
 @sphinx_intl
@@ -875,15 +1012,16 @@ def test_html_rebuild_mo(app):
     app.build()
     # --- rebuild by .mo mtime
     app.build()
-    app.env.find_files(app.config, app.builder)
-    _, updated, _ = app.env.get_outdated_files(config_changed=False)
-    assert len(updated) == 0
+    _, updated, _ = _get_update_targets(app)
+    assert updated == set()
 
-    mtime = (app.srcdir / 'xx' / 'LC_MESSAGES' / 'bom.mo').stat().st_mtime
-    os.utime(app.srcdir / 'xx' / 'LC_MESSAGES' / 'bom.mo', (mtime + 5, mtime + 5))
-    app.env.find_files(app.config, app.builder)
-    _, updated, _ = app.env.get_outdated_files(config_changed=False)
-    assert len(updated) == 1
+    _, bom_file = _get_bom_intl_path(app)
+    old_mtime = bom_file.stat().st_mtime
+    new_mtime = old_mtime + (dt := 5)
+    os.utime(bom_file, (new_mtime, new_mtime))
+    assert old_mtime + dt == new_mtime, (old_mtime + dt, new_mtime)
+    _, updated, _ = _get_update_targets(app)
+    assert updated == {'bom'}
 
 
 @sphinx_intl
@@ -1222,7 +1360,7 @@ def test_additional_targets_should_not_be_translated(app):
     'html',
     srcdir='test_additional_targets_should_be_translated',
     confoverrides={
-        'language': 'xx', 'locale_dirs': ['.'],
+        'language': _CATALOG_LOCALE, 'locale_dirs': ['.'],
         'gettext_compact': False,
         'gettext_additional_targets': [
             'index',
@@ -1300,7 +1438,7 @@ def test_additional_targets_should_be_translated(app):
     'html',
     testroot='intl_substitution_definitions',
     confoverrides={
-        'language': 'xx', 'locale_dirs': ['.'],
+        'language': _CATALOG_LOCALE, 'locale_dirs': ['.'],
         'gettext_compact': False,
         'gettext_additional_targets': [
             'index',
@@ -1342,7 +1480,7 @@ def test_text_references(app, warning):
     'text',
     testroot='intl_substitution_definitions',
     confoverrides={
-        'language': 'xx', 'locale_dirs': ['.'],
+        'language': _CATALOG_LOCALE, 'locale_dirs': ['.'],
         'gettext_compact': False,
     },
 )
@@ -1368,7 +1506,7 @@ SUBSTITUTED IMAGE [image: SUBST_EPILOG_2 TRANSLATED][image] HERE.
 @pytest.mark.sphinx(
     'dummy', testroot='images',
     srcdir='test_intl_images',
-    confoverrides={'language': 'xx'},
+    confoverrides={'language': _CATALOG_LOCALE},
 )
 def test_image_glob_intl(app):
     app.build()
@@ -1412,7 +1550,7 @@ def test_image_glob_intl(app):
     'dummy', testroot='images',
     srcdir='test_intl_images',
     confoverrides={
-        'language': 'xx',
+        'language': _CATALOG_LOCALE,
         'figure_language_filename': '{root}{ext}.{language}',
     },
 )