python-fluent/fluent.runtime/tests/test_fallback.py at 1656a3551daed836ec306a91ba3182eb63813df4 · projectfluent/python-fluent · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
from os.path import join

from fluent.runtime import FluentLocalization, FluentResourceLoader

from .utils import build_file_tree


class TestLocalization:
    def test_init(self):
        l10n = FluentLocalization(
            ["en"], ["file.ftl"], FluentResourceLoader("{locale}")
        )
        assert callable(l10n.format_value)

    def test_bundles(self, tmp_path):
        build_file_tree(
            tmp_path,
            {
                "de": {
                    "one.ftl": "one = in German\n  .foo = one in German\n",
                    "two.ftl": "two = in German\n  .foo = two in German\n",
                },
                "fr": {"two.ftl": "three = in French\n  .foo = three in French\n"},
                "en": {
                    "one.ftl": "four = exists\n  .foo = four in English\n",
                    "two.ftl": "five = exists\n  .foo = five in English\n"
                    + "bar =\n  .foo = bar in English\n"
                    + "baz = baz in English\n",
                },
            },
        )
        l10n = FluentLocalization(
            ["de", "fr", "en"],
            ["one.ftl", "two.ftl"],
            FluentResourceLoader(join(tmp_path, "{locale}")),
        )
        bundles_gen = l10n._bundles()
        bundle_de = next(bundles_gen)
        assert bundle_de.locales[0] == "de"
        assert bundle_de.has_message("one")
        assert bundle_de.has_message("two")
        bundle_fr = next(bundles_gen)
        assert bundle_fr.locales[0] == "fr"
        assert not bundle_fr.has_message("one")
        assert bundle_fr.has_message("three")
        assert list(l10n._bundles())[:2] == [bundle_de, bundle_fr]
        bundle_en = next(bundles_gen)
        assert bundle_en.locales[0] == "en"
        assert l10n.format_value("one") == "in German"
        assert l10n.format_value("two") == "in German"
        assert l10n.format_value("three") == "in French"
        assert l10n.format_value("four") == "exists"
        assert l10n.format_value("five") == "exists"
        assert l10n.format_value("bar") == "bar"
        assert l10n.format_value("baz") == "baz in English"
        assert l10n.format_value("not-exists") == "not-exists"
        assert tuple(l10n.format_message("one")) == (
            "in German",
            {"foo": "one in German"},
        )
        assert tuple(l10n.format_message("two")) == (
            "in German",
            {"foo": "two in German"},
        )
        assert tuple(l10n.format_message("three")) == (
            "in French",
            {"foo": "three in French"},
        )
        assert tuple(l10n.format_message("four")) == (
            "exists",
            {"foo": "four in English"},
        )
        assert tuple(l10n.format_message("five")) == (
            "exists",
            {"foo": "five in English"},
        )
        assert tuple(l10n.format_message("bar")) == (None, {"foo": "bar in English"})
        assert tuple(l10n.format_message("baz")) == ("baz in English", {})
        assert tuple(l10n.format_message("not-exists")) == ("not-exists", {})

    def test_format_value_is_thread_safe(self, tmp_path):
        # Regression test for issue #221: concurrent format_value() calls used
        # to race on the shared _bundle_it generator and raise
        # "ValueError: generator already executing".
        import threading

        build_file_tree(
            tmp_path,
            {
                "en": {"one.ftl": "hello = world\n"},
            },
        )
        l10n = FluentLocalization(
            ["en"], ["one.ftl"], FluentResourceLoader(join(tmp_path, "{locale}"))
        )
        errors: list[BaseException] = []
        start = threading.Event()

        def worker():
            start.wait()
            try:
                for _ in range(50):
                    assert l10n.format_value("hello") == "world"
            except BaseException as exc:
                errors.append(exc)

        threads = [threading.Thread(target=worker) for _ in range(8)]
        for t in threads:
            t.start()
        start.set()
        for t in threads:
            t.join()
        assert errors == []


class TestResourceLoader:
    def test_all_exist(self, tmp_path):
        build_file_tree(
            tmp_path,
            {
                "en": {
                    "one.ftl": "one = exists",
                    "two.ftl": "two = exists",
                }
            },
        )
        loader = FluentResourceLoader(join(tmp_path, "{locale}"))
        resources_list = list(loader.resources("en", ["one.ftl", "two.ftl"]))
        assert len(resources_list) == 1
        resources = resources_list[0]
        assert len(resources) == 2

    def test_one_exists(self, tmp_path):
        build_file_tree(tmp_path, {"en": {"two.ftl": "two = exists"}})
        loader = FluentResourceLoader(join(tmp_path, "{locale}"))
        resources_list = list(loader.resources("en", ["one.ftl", "two.ftl"]))
        assert len(resources_list) == 1
        resources = resources_list[0]
        assert len(resources) == 1

    def test_none_exist(self, tmp_path):
        loader = FluentResourceLoader(join(tmp_path, "{locale}"))
        resources_list = list(loader.resources("en", ["one.ftl", "two.ftl"]))
        assert len(resources_list) == 0