|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154 |
- import unittest
- from unittest.mock import patch, MagicMock
- import requests
- from services.headlines import prepare_headlines, is_headline, DEFAULT_TIMEOUT, DEFAULT_HEADERS
-
-
- class TestHeadlinesTimeout(unittest.TestCase):
- def setUp(self):
- self.stopwords = {"the", "a", "an", "in", "on"}
-
- @patch("services.headlines.requests.get")
- def test_default_timeout_used(self, mock_get):
- mock_response = MagicMock()
- mock_response.status_code = 200
- mock_response.content = b"<html><body><a href='#'>Default Timeout Headline</a></body></html>"
- mock_response.text = "<html><body><a href='#'>Default Timeout Headline</a></body></html>"
- mock_get.return_value = mock_response
-
- sources = ["https://example.com/news"]
- headlines = prepare_headlines(sources, self.stopwords)
-
- mock_get.assert_called_once_with("https://example.com/news", allow_redirects=True, timeout=DEFAULT_TIMEOUT, headers=DEFAULT_HEADERS)
- self.assertEqual(len(headlines), 1)
- self.assertEqual(headlines[0].display_text, "Default Timeout Headline")
-
- @patch("services.headlines.requests.get")
- def test_custom_timeout_used(self, mock_get):
- mock_response = MagicMock()
- mock_response.status_code = 200
- mock_response.content = b"<html><body><a href='#'>Custom Timeout Headline</a></body></html>"
- mock_response.text = "<html><body><a href='#'>Custom Timeout Headline</a></body></html>"
- mock_get.return_value = mock_response
-
- sources = ["https://example.com/news"]
- headlines = prepare_headlines(sources, self.stopwords, timeout=10)
-
- mock_get.assert_called_once_with("https://example.com/news", allow_redirects=True, timeout=10, headers=DEFAULT_HEADERS)
- self.assertEqual(len(headlines), 1)
-
- @patch("services.headlines.requests.get")
- def test_timeout_skips_slow_request_and_processes_others(self, mock_get):
- slow_url = "https://slow-source.example.com"
- fast_url = "https://fast-source.example.com"
-
- def side_effect(url, **kwargs):
- if url == slow_url:
- raise requests.exceptions.Timeout("Connection timed out after %s seconds" % kwargs.get("timeout"))
- fast_response = MagicMock()
- fast_response.status_code = 200
- fast_response.content = b"<html><body><span>Breaking News Story</span></body></html>"
- fast_response.text = "<html><body><span>Breaking News Story</span></body></html>"
- return fast_response
-
- mock_get.side_effect = side_effect
-
- sources = [slow_url, fast_url]
- headlines = prepare_headlines(sources, self.stopwords, timeout=3)
-
- self.assertEqual(mock_get.call_count, 2)
- self.assertEqual(len(headlines), 1)
- self.assertEqual(headlines[0].display_text, "Breaking News Story")
-
- @patch("services.headlines.requests.get")
- def test_connect_timeout_and_read_timeout_skipped(self, mock_get):
- connect_timeout_url = "https://connect-timeout.com"
- read_timeout_url = "https://read-timeout.com"
-
- mock_get.side_effect = [
- requests.exceptions.ConnectTimeout("Connect timeout"),
- requests.exceptions.ReadTimeout("Read timeout")
- ]
-
- sources = [connect_timeout_url, read_timeout_url]
- headlines = prepare_headlines(sources, self.stopwords)
-
- self.assertEqual(mock_get.call_count, 2)
- self.assertEqual(len(headlines), 0)
-
-
- class TestHeadlineSafeguards(unittest.TestCase):
- def setUp(self):
- self.stopwords = {"the", "a", "an", "in", "on", "and", "of", "to"}
-
- def test_non_headline_link_texts_rejected(self):
- non_headlines = [
- "Account Settings",
- "Follow",
- "Television",
- "Sign In",
- "Home",
- "Politics",
- "About Us",
- "Contact Us",
- "Menu",
- "Search",
- "",
- " ",
- "123",
- "...",
- "in on a", # only stopwords
- ]
- for item in non_headlines:
- with self.subTest(item=item):
- self.assertFalse(is_headline(item, self.stopwords), f"{item!r} should not be recognized as a headline")
-
- def test_valid_headlines_accepted(self):
- valid_headlines = [
- "Breaking News Story",
- "Custom Timeout Headline",
- "Senate passes major infrastructure bill",
- "Scientists discover new ocean species",
- "Federal Reserve holds interest rates steady",
- ]
- for item in valid_headlines:
- with self.subTest(item=item):
- self.assertTrue(is_headline(item, self.stopwords), f"{item!r} should be recognized as a headline")
-
- @patch("services.headlines.requests.get")
- def test_prepare_headlines_filters_out_navigation_and_non_headlines(self, mock_get):
- html_content = """
- <html>
- <body>
- <a href="/settings">Account Settings</a>
- <a href="/social">Follow</a>
- <a href="/tv">Television</a>
- <a href="/login">Sign In</a>
- <a href="/article1">Senate passes major infrastructure bill</a>
- <span>Home</span>
- <span>Federal Reserve holds interest rates steady</span>
- </body>
- </html>
- """
- mock_response = MagicMock()
- mock_response.status_code = 200
- mock_response.content = html_content.encode("utf-8")
- mock_response.text = html_content
- mock_get.return_value = mock_response
-
- sources = ["https://example.com/news"]
- headlines = prepare_headlines(sources, self.stopwords)
-
- self.assertEqual(len(headlines), 2)
- extracted_texts = [h.display_text for h in headlines]
- self.assertIn("Senate passes major infrastructure bill", extracted_texts)
- self.assertIn("Federal Reserve holds interest rates steady", extracted_texts)
- self.assertNotIn("Account Settings", extracted_texts)
- self.assertNotIn("Follow", extracted_texts)
- self.assertNotIn("Television", extracted_texts)
- self.assertNotIn("Sign In", extracted_texts)
- self.assertNotIn("Home", extracted_texts)
-
-
- if __name__ == "__main__":
- unittest.main()
|