68 lines
3.5 KiB
Python
68 lines
3.5 KiB
Python
from argparse import Namespace
|
|
import io
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from car_tracker.collect import collect_browser
|
|
|
|
|
|
class BrowserCleanupTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.path = Path(temporary.name) / "manifest.json"
|
|
self.path.write_text('{"status":"complete"}')
|
|
self.args = Namespace(headed=False, searches=[], output=self.path.parent,
|
|
max_pages=1, delay=1, search_only=True, max_details=None)
|
|
self.playwright = self.enterContext(patch("playwright.sync_api.sync_playwright"))
|
|
self.stealth = self.enterContext(patch("playwright_stealth.Stealth"))
|
|
self.stealth.return_value.use_sync.side_effect = lambda manager: manager
|
|
self.browser = self.playwright.return_value.__enter__.return_value.chromium.launch.return_value
|
|
self.collect = self.enterContext(patch("car_tracker.collect.collect", return_value=(self.path, "complete")))
|
|
self.stderr = self.enterContext(patch("sys.stderr", new_callable=io.StringIO))
|
|
|
|
def test_stealth_wraps_playwright_with_french_languages_in_both_modes(self):
|
|
for headed in (False, True):
|
|
with self.subTest(headed=headed):
|
|
self.args.headed = headed
|
|
self.stealth.reset_mock()
|
|
self.browser.reset_mock()
|
|
self.assertEqual(collect_browser(self.args), (self.path, "complete"))
|
|
self.stealth.assert_called_once_with(navigator_languages_override=("fr-FR", "fr"))
|
|
self.stealth.return_value.use_sync.assert_called_once_with(self.playwright.return_value)
|
|
self.playwright.return_value.__enter__.return_value.chromium.launch.assert_called_with(
|
|
headless=not headed,
|
|
)
|
|
self.browser.new_context.assert_called_once_with(locale="fr-FR")
|
|
self.browser.new_context.return_value.new_page.assert_called_once_with()
|
|
self.browser.close.assert_called_once_with()
|
|
|
|
def test_stealth_setup_failure_is_not_suppressed(self):
|
|
self.stealth.return_value.use_sync.side_effect = RuntimeError("stealth failed")
|
|
with self.assertRaisesRegex(RuntimeError, "stealth failed"):
|
|
collect_browser(self.args)
|
|
self.collect.assert_not_called()
|
|
|
|
def test_browser_close_failure_retains_finalized_result(self):
|
|
self.browser.close.side_effect = RuntimeError("SYNTHETIC_SECRET")
|
|
self.assertEqual(collect_browser(self.args), (self.path, "complete"))
|
|
self.assertTrue((self.path.parent / "browser_cleanup.json").exists())
|
|
self.assertNotIn("SYNTHETIC_SECRET", self.stderr.getvalue())
|
|
|
|
def test_driver_cleanup_failure_retains_finalized_result(self):
|
|
self.playwright.return_value.__exit__.side_effect = RuntimeError("cleanup failed")
|
|
self.assertEqual(collect_browser(self.args), (self.path, "complete"))
|
|
|
|
def test_collection_failure_is_not_suppressed(self):
|
|
self.collect.side_effect = OSError("write failed")
|
|
with self.assertRaises(OSError):
|
|
collect_browser(self.args)
|
|
|
|
def test_cleanup_interruption_remains_cancellation_with_manifest_path(self):
|
|
self.browser.close.side_effect = KeyboardInterrupt
|
|
with self.assertRaises(KeyboardInterrupt):
|
|
collect_browser(self.args)
|
|
self.assertIn(str(self.path), self.stderr.getvalue())
|