Kevin Veen-Birkenbach
747ce379cc
ProxyFix defaults x_for to 1, so ProxyFix(app.wsgi_app, x_proto=1) never disabled it: request.remote_addr and the access log were forgeable by any client that reached the app directly. It is x_for=0 now, asserted rather than assumed. A mutation audit over the change set reverted 196 deliberate behaviours and found 47 that no test noticed. This closes the ones that carry damage: - apod_background lost its key check, its transport guard, its status guard and its media-type check without a single test failing. Each one turns a slow or unhappy NASA into a 500 on every page. - Untrusted values reached innerHTML through window.I18N, which the translation backend writes, and the modal's click handlers stacked so a later click opened an earlier popup's URL. - The sync tool could ask for HTML instead of text, translate from "auto" instead of English, run without a timeout, store an empty translation that marks the string done for good, abandon 28 languages because one could not be written, and report success after reaching nothing. - Neither the lint target, the CI jobs, the vendored RTL stylesheet, the documented environment keys, nor any of the four hardenings in scripts/run-e2e.sh was observed by anything. Three of the new tests passed for the wrong reason on their first cut — a mock that answered None whether or not the guard existed, a raise_for_status that was never called, a string that stayed in the file after the mutation. The audit found those too; all 24 reverts now fail. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
358 lines
13 KiB
Python
358 lines
13 KiB
Python
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import yaml
|
|
|
|
from app.utils import i18n
|
|
from utils import i18n_sync
|
|
|
|
|
|
class TestCollectSources(unittest.TestCase):
|
|
def test_prose_keys_are_collected_from_nested_structures(self):
|
|
config = {
|
|
"cards": [
|
|
{"title": "Agile Coach", "text": "I lead transformations."},
|
|
{"text": "Another card."},
|
|
],
|
|
"navigation": {
|
|
"header": {
|
|
"children": [
|
|
{"name": "Apps", "description": "Application menu"},
|
|
]
|
|
}
|
|
},
|
|
"platform": {"titel": "Someone", "subtitel": "A tagline"},
|
|
}
|
|
|
|
self.assertEqual(
|
|
i18n_sync.collect_sources(config),
|
|
{
|
|
"I lead transformations.",
|
|
"Another card.",
|
|
"Application menu",
|
|
"A tagline",
|
|
},
|
|
)
|
|
|
|
def test_label_and_structural_keys_are_left_alone(self):
|
|
config = {
|
|
"name": "Mastodon",
|
|
"title": "Cybermaster",
|
|
"url": "https://example.test",
|
|
"link_text": "www.example.test",
|
|
}
|
|
|
|
self.assertEqual(i18n_sync.collect_sources(config), set())
|
|
|
|
def test_blank_values_are_ignored(self):
|
|
self.assertEqual(i18n_sync.collect_sources({"text": " "}), set())
|
|
|
|
def test_a_list_of_prose_is_collected(self):
|
|
self.assertEqual(
|
|
i18n_sync.collect_sources({"text": ["one", "two"]}), {"one", "two"}
|
|
)
|
|
|
|
|
|
class TestTranslate(unittest.TestCase):
|
|
def _session(self, ok, payload=None, status=200, text=""):
|
|
response = Mock(ok=ok, status_code=status, text=text)
|
|
response.json.return_value = payload or {}
|
|
return Mock(post=Mock(return_value=response))
|
|
|
|
def test_successful_response_returns_the_translation(self):
|
|
session = self._session(True, {"translatedText": "Hallo"})
|
|
|
|
result = i18n_sync.translate(session, "http://lt", "", "Hello", "de")
|
|
|
|
self.assertEqual(result, "Hallo")
|
|
|
|
def test_api_key_is_only_sent_when_configured(self):
|
|
session = self._session(True, {"translatedText": "Hallo"})
|
|
|
|
i18n_sync.translate(session, "http://lt", "secret", "Hello", "de")
|
|
|
|
self.assertEqual(session.post.call_args.kwargs["data"]["api_key"], "secret")
|
|
|
|
def test_no_api_key_is_sent_when_none_is_configured(self):
|
|
session = self._session(True, {"translatedText": "Hallo"})
|
|
|
|
i18n_sync.translate(session, "http://lt", "", "Hello", "de")
|
|
|
|
self.assertNotIn("api_key", session.post.call_args.kwargs["data"])
|
|
|
|
def test_failed_response_returns_none(self):
|
|
session = self._session(
|
|
False, {"translatedText": "SHOULD NOT BE USED"}, status=403, text="denied"
|
|
)
|
|
|
|
self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de"))
|
|
|
|
def test_the_request_asks_for_plain_text_from_the_source_language(self):
|
|
session = self._session(True, {"translatedText": "Hallo"})
|
|
|
|
i18n_sync.translate(session, "http://lt", "", "Hello", "de")
|
|
|
|
data = session.post.call_args.kwargs["data"]
|
|
self.assertEqual(data["format"], "text")
|
|
self.assertEqual(data["source"], i18n.SOURCE_LANGUAGE)
|
|
self.assertEqual(data["target"], "de")
|
|
|
|
def test_the_request_is_bounded_by_a_timeout(self):
|
|
session = self._session(True, {"translatedText": "Hallo"})
|
|
|
|
i18n_sync.translate(session, "http://lt", "", "Hello", "de")
|
|
|
|
timeout = session.post.call_args.kwargs["timeout"]
|
|
self.assertIsInstance(timeout, (int, float))
|
|
self.assertGreater(timeout, 0)
|
|
|
|
def test_an_empty_translation_is_refused(self):
|
|
session = self._session(True, {"translatedText": ""})
|
|
|
|
self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de"))
|
|
|
|
def test_a_non_json_success_response_returns_none(self):
|
|
response = Mock(ok=True)
|
|
response.json.side_effect = ValueError("no json")
|
|
session = Mock(post=Mock(return_value=response))
|
|
|
|
self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de"))
|
|
|
|
def test_a_transport_failure_returns_none(self):
|
|
session = Mock(
|
|
post=Mock(side_effect=i18n_sync.requests.ConnectionError("reset"))
|
|
)
|
|
|
|
self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de"))
|
|
|
|
def test_a_non_string_translation_is_refused(self):
|
|
session = self._session(True, {"translatedText": ["Hallo", "Welt"]})
|
|
|
|
self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de"))
|
|
|
|
|
|
class TestSupportedLanguages(unittest.TestCase):
|
|
def _session(self, payload=None, side_effect=None):
|
|
response = Mock(raise_for_status=Mock())
|
|
if side_effect is not None:
|
|
response.json.side_effect = side_effect
|
|
else:
|
|
response.json.return_value = payload
|
|
return Mock(get=Mock(return_value=response))
|
|
|
|
def test_the_offered_codes_are_returned(self):
|
|
session = self._session([{"code": "de"}, {"code": "fr"}])
|
|
|
|
self.assertEqual(
|
|
i18n_sync.supported_languages("http://lt", session), {"de", "fr"}
|
|
)
|
|
|
|
def test_an_unreachable_instance_is_reported(self):
|
|
session = Mock(get=Mock(side_effect=i18n_sync.requests.ConnectTimeout("slow")))
|
|
|
|
with self.assertRaises(i18n_sync.BackendError):
|
|
i18n_sync.supported_languages("http://lt", session)
|
|
|
|
def test_an_error_status_is_reported(self):
|
|
response = Mock()
|
|
response.raise_for_status.side_effect = i18n_sync.requests.HTTPError("503")
|
|
response.json.return_value = [{"code": "de"}]
|
|
session = Mock(get=Mock(return_value=response))
|
|
|
|
with self.assertRaises(i18n_sync.BackendError):
|
|
i18n_sync.supported_languages("http://lt", session)
|
|
|
|
def test_a_non_json_listing_is_reported(self):
|
|
session = self._session(side_effect=ValueError("no json"))
|
|
|
|
with self.assertRaises(i18n_sync.BackendError):
|
|
i18n_sync.supported_languages("http://lt", session)
|
|
|
|
def test_json_of_the_wrong_shape_is_reported(self):
|
|
for payload in ({"error": "Slow down"}, ["de", "fr"], None, [{"name": "de"}]):
|
|
with self.subTest(payload=payload):
|
|
with self.assertRaises(i18n_sync.BackendError):
|
|
i18n_sync.supported_languages("http://lt", self._session(payload))
|
|
|
|
|
|
class TestLoadExisting(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, self.directory, True)
|
|
self.path = self.directory / "de.yaml"
|
|
|
|
def test_a_missing_file_is_an_empty_catalog(self):
|
|
self.assertEqual(i18n_sync.load_existing(self.path), {})
|
|
|
|
def test_a_readable_catalog_is_returned(self):
|
|
self.path.write_text("Hi: Hallo\n", encoding="utf-8")
|
|
|
|
self.assertEqual(i18n_sync.load_existing(self.path), {"Hi": "Hallo"})
|
|
|
|
def test_an_unparsable_catalog_is_refused(self):
|
|
self.path.write_text('Hi: "unterminated\n', encoding="utf-8")
|
|
|
|
self.assertIsNone(i18n_sync.load_existing(self.path))
|
|
|
|
def test_a_non_mapping_catalog_is_refused(self):
|
|
self.path.write_text("- a\n- b\n", encoding="utf-8")
|
|
|
|
self.assertIsNone(i18n_sync.load_existing(self.path))
|
|
|
|
def test_a_non_utf8_catalog_is_refused(self):
|
|
self.path.write_bytes(b"\xffHi: Hallo\n")
|
|
|
|
self.assertIsNone(i18n_sync.load_existing(self.path))
|
|
|
|
def test_a_directory_at_the_catalog_path_is_refused(self):
|
|
(self.directory / "sub.yaml").mkdir()
|
|
|
|
self.assertIsNone(i18n_sync.load_existing(self.directory / "sub.yaml"))
|
|
|
|
def test_an_empty_or_comment_only_catalog_is_an_empty_catalog(self):
|
|
for text in ("", "\n\n", "# only a comment\n"):
|
|
with self.subTest(text=text):
|
|
self.path.write_text(text, encoding="utf-8")
|
|
self.assertEqual(i18n_sync.load_existing(self.path), {})
|
|
|
|
|
|
class TestWriteCatalog(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, self.directory, True)
|
|
self.path = self.directory / "de.yaml"
|
|
|
|
def test_the_catalog_is_written_and_no_scratch_file_is_left(self):
|
|
i18n_sync.write_catalog(self.path, {"Hi": "Hallo", "Umlaut": "Grüße"})
|
|
|
|
text = self.path.read_text(encoding="utf-8")
|
|
self.assertEqual(yaml.safe_load(text), {"Hi": "Hallo", "Umlaut": "Grüße"})
|
|
self.assertIn("Grüße", text)
|
|
self.assertEqual([p.name for p in self.directory.iterdir()], ["de.yaml"])
|
|
|
|
def test_a_write_that_dies_halfway_leaves_the_previous_catalog_intact(self):
|
|
self.path.write_text("Hi: HAND-EDITED\n", encoding="utf-8")
|
|
complete = Path.write_text
|
|
|
|
def dies_halfway(target, data, *args, **kwargs):
|
|
complete(target, data[: len(data) // 2], *args, **kwargs)
|
|
raise OSError("no space left on device")
|
|
|
|
with patch.object(Path, "write_text", dies_halfway):
|
|
with self.assertRaises(OSError):
|
|
i18n_sync.write_catalog(self.path, {"Hi": "machine", "Zebra": "Zebra"})
|
|
|
|
self.assertEqual(self.path.read_text(encoding="utf-8"), "Hi: HAND-EDITED\n")
|
|
|
|
|
|
class TestSync(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, self.directory, True)
|
|
self.addCleanup(setattr, i18n, "CONTENT_DIR", i18n.CONTENT_DIR)
|
|
i18n.CONTENT_DIR = self.directory
|
|
self.path = self.directory / "de.yaml"
|
|
|
|
def _run(self, sources, offered=("de",), translated="UEBERSETZT"):
|
|
listing = Mock(raise_for_status=Mock())
|
|
listing.json.return_value = [{"code": code} for code in offered]
|
|
answer = Mock(ok=True)
|
|
answer.json.return_value = {"translatedText": translated}
|
|
session = Mock(get=Mock(return_value=listing), post=Mock(return_value=answer))
|
|
|
|
with patch.object(i18n_sync.requests, "Session", return_value=session):
|
|
i18n_sync.sync("http://lt", "", set(sources), ["de"], self.directory)
|
|
return session
|
|
|
|
def test_missing_entries_are_written(self):
|
|
self._run(["Hello"])
|
|
|
|
self.assertEqual(
|
|
yaml.safe_load(self.path.read_text(encoding="utf-8")),
|
|
{"Hello": "UEBERSETZT"},
|
|
)
|
|
|
|
def test_existing_entries_are_never_overwritten(self):
|
|
self.path.write_text("Hello: HAND-EDITED\n", encoding="utf-8")
|
|
|
|
session = self._run(["Hello", "World"])
|
|
|
|
catalog = yaml.safe_load(self.path.read_text(encoding="utf-8"))
|
|
self.assertEqual(catalog["Hello"], "HAND-EDITED")
|
|
self.assertEqual(session.post.call_count, 1)
|
|
|
|
def test_an_unparsable_catalog_is_left_alone(self):
|
|
broken = 'Hello: "unterminated\n'
|
|
self.path.write_text(broken, encoding="utf-8")
|
|
|
|
session = self._run(["Hello"])
|
|
|
|
self.assertEqual(self.path.read_text(encoding="utf-8"), broken)
|
|
session.post.assert_not_called()
|
|
|
|
def test_a_language_the_instance_does_not_offer_is_skipped(self):
|
|
session = self._run(["Hello"], offered=("fr",))
|
|
|
|
self.assertFalse(self.path.exists())
|
|
session.post.assert_not_called()
|
|
|
|
def test_an_empty_translation_is_not_stored(self):
|
|
self._run(["Hello"], translated="")
|
|
|
|
self.assertFalse(self.path.exists())
|
|
|
|
def test_a_run_that_translated_nothing_leaves_the_file_untouched(self):
|
|
original = "# Reviewed by a native speaker, keep the order.\nZebra: Zebra\n"
|
|
self.path.write_text(original, encoding="utf-8")
|
|
|
|
self._run(["Hello"], translated=None)
|
|
|
|
self.assertEqual(self.path.read_text(encoding="utf-8"), original)
|
|
|
|
def test_a_string_the_shipped_catalog_covers_is_not_requested(self):
|
|
ui = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, ui, True)
|
|
self.addCleanup(setattr, i18n, "UI_DIR", i18n.UI_DIR)
|
|
i18n.UI_DIR = ui
|
|
(ui / "de.yaml").write_text("Close: Schliessen\n", encoding="utf-8")
|
|
|
|
session = self._run(["Close", "Hello"])
|
|
|
|
self.assertEqual(session.post.call_count, 1)
|
|
self.assertNotIn("Close", yaml.safe_load(self.path.read_text(encoding="utf-8")))
|
|
|
|
def test_a_language_that_cannot_be_written_does_not_stop_the_others(self):
|
|
listing = Mock(raise_for_status=Mock())
|
|
listing.json.return_value = [{"code": "de"}, {"code": "fr"}]
|
|
answer = Mock(ok=True)
|
|
answer.json.return_value = {"translatedText": "UEBERSETZT"}
|
|
session = Mock(get=Mock(return_value=listing), post=Mock(return_value=answer))
|
|
original = i18n_sync.write_catalog
|
|
|
|
def fails_for_german(path, catalog):
|
|
if path.name == "de.yaml":
|
|
raise OSError("read-only")
|
|
original(path, catalog)
|
|
|
|
with patch.object(i18n_sync, "write_catalog", fails_for_german):
|
|
with patch.object(i18n_sync.requests, "Session", return_value=session):
|
|
i18n_sync.sync("http://lt", "", {"Hello"}, ["de", "fr"], self.directory)
|
|
|
|
self.assertTrue((self.directory / "fr.yaml").is_file())
|
|
|
|
def test_the_catalog_directory_is_created(self):
|
|
nested = self.directory / "content"
|
|
self.directory = nested
|
|
self.path = nested / "de.yaml"
|
|
|
|
self._run(["Hello"])
|
|
|
|
self.assertTrue(self.path.is_file())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|