From a7593c34d32ede15b8136a0d40aa80aeb0b79fa5 Mon Sep 17 00:00:00 2001 From: Florian S Date: Sat, 25 Jan 2025 18:53:16 +0100 Subject: [PATCH 1/4] Allow to pass a requests.Session object to the miniflux client. This update introduces the ability to pass a `requests.Session` object to the `Miniflux` client. This change enables users to leverage session management features such as rate limiting and caching. **Changes:** - Added `session: requests.Session` as an optional parameter to the `Miniflux` class constructor. - Update the constructor to configure the session and trying to avoid to clash with any preset Session config - The session is used for every HTTP request - The default behavior creates a new session for each client instance. **Example (header configuration):** ```python import requests import miniflux session = requests.Session() session.headers.update({"User-Agent": "Custom User-Agent"}) client = miniflux.Client("https://reader.miniflux.app", session=session) ``` **Example (client-side caching):** ```python import miniflux from cachecontrol import CacheControl from cachecontrol.heuristics import ExpiresAfter session = CacheControl(requests.Session(), heuristic=ExpiresAfter(days=1)) client = miniflux.Client("https://reader.miniflux.app", session=session) ``` **Possible improvement:** The `request.Session` object is auto-closeable, would it make sense to make the `miniflux.Client` auto-closeable as well to close the session ? Let me know if you have any questions or suggestions! --- miniflux.py | 193 ++++++------- tests/test_client.py | 637 ++++++++++++++++++++----------------------- 2 files changed, 373 insertions(+), 457 deletions(-) diff --git a/miniflux.py b/miniflux.py index c4a7173..e9ce552 100644 --- a/miniflux.py +++ b/miniflux.py @@ -110,18 +110,21 @@ def __init__( timeout: float = 30.0, api_key: Optional[str] = None, user_agent: str = DEFAULT_USER_AGENT, + session: requests.Session = requests.Session() ): self._base_url = base_url - self._api_key = api_key - self._username = username - self._password = password self._timeout = timeout - self._auth: Optional[tuple] = ( - (self._username, self._password) if not api_key else None + self._session = session + + auth : Optional[tuple] = ( + (username or "", password or "") if not api_key else None ) - self._headers = {"User-Agent": user_agent} + + self._session.headers.update({"User-Agent": user_agent}) if api_key: - self._headers["X-Auth-Token"] = api_key + self._session.headers.update({"X-Auth-Token": api_key}) + if auth != None: + self._session.auth = auth def _get_endpoint(self, path: str) -> str: if len(self._base_url) > 0 and self._base_url[-1:] == "/": @@ -157,8 +160,8 @@ def flush_history(self) -> bool: bool: True if the operation was successfully scheduled, False otherwise. """ endpoint = self._get_endpoint("/flush-history") - response = requests.delete( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.delete( + endpoint, timeout=self._timeout ) if response.status_code == 202: return True @@ -174,8 +177,8 @@ def get_version(self) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint("/version") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -191,8 +194,8 @@ def me(self) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint("/me") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -219,8 +222,8 @@ def export_feeds(self) -> str: ClientError: If the request fails. """ endpoint = self._get_endpoint("/export") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.text @@ -238,11 +241,9 @@ def import_feeds(self, opml: str) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint("/import") - response = requests.post( + response = self._session.post( endpoint, - headers=self._headers, data=opml, - auth=self._auth, timeout=self._timeout, ) if response.status_code == 201: @@ -264,10 +265,8 @@ def discover(self, website_url: str, **kwargs) -> List[dict]: data = dict(url=website_url) data.update(kwargs) - response = requests.post( + response = self._session.post( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -287,8 +286,8 @@ def get_category_feeds(self, category_id: int) -> List[dict]: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/categories/{category_id}/feeds") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -304,8 +303,8 @@ def get_feeds(self) -> List[dict]: ClientError: If the request fails. """ endpoint = self._get_endpoint("/feeds") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -323,8 +322,8 @@ def get_feed(self, feed_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/feeds/{feed_id}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -342,8 +341,8 @@ def get_feed_icon(self, feed_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/feeds/{feed_id}/icon") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -361,8 +360,8 @@ def get_icon(self, icon_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/icons/{icon_id}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -399,10 +398,8 @@ def create_feed( data = dict(feed_url=feed_url, category_id=category_id) data.update(kwargs) - response = requests.post( + response = self._session.post( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -423,10 +420,8 @@ def update_feed(self, feed_id: int, **kwargs) -> dict: """ endpoint = self._get_endpoint(f"/feeds/{feed_id}") data = self._get_modification_params(**kwargs) - response = requests.put( + response = self._session.put( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -444,8 +439,8 @@ def refresh_all_feeds(self) -> bool: ClientError: If the request fails. """ endpoint = self._get_endpoint("/feeds/refresh") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code >= 400: self._handle_error_response(response) @@ -463,8 +458,8 @@ def refresh_feed(self, feed_id: int) -> bool: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/feeds/{feed_id}/refresh") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code >= 400: self._handle_error_response(response) @@ -482,8 +477,8 @@ def refresh_category(self, category_id: int) -> bool: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/categories/{category_id}/refresh") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code >= 400: self._handle_error_response(response) @@ -499,8 +494,8 @@ def delete_feed(self, feed_id: int) -> None: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/feeds/{feed_id}") - response = requests.delete( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.delete( + endpoint, timeout=self._timeout ) if response.status_code != 204: self._handle_error_response(response) @@ -518,8 +513,8 @@ def get_feed_entry(self, feed_id: int, entry_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/feeds/{feed_id}/entries/{entry_id}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -538,10 +533,8 @@ def get_feed_entries(self, feed_id: int, **kwargs) -> dict: """ endpoint = self._get_endpoint(f"/feeds/{feed_id}/entries") params = self._get_params(**kwargs) - response = requests.get( + response = self._session.get( endpoint, - headers=self._headers, - auth=self._auth, params=params, timeout=self._timeout, ) @@ -561,8 +554,8 @@ def mark_feed_entries_as_read(self, feed_id: int) -> None: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/feeds/{feed_id}/mark-all-as-read") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code != 204: self._handle_error_response(response) @@ -579,8 +572,8 @@ def get_entry(self, entry_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/entries/{entry_id}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -597,10 +590,8 @@ def get_entries(self, **kwargs) -> dict: """ endpoint = self._get_endpoint("/entries") params = self._get_params(**kwargs) - response = requests.get( + response = self._session.get( endpoint, - headers=self._headers, - auth=self._auth, params=params, timeout=self._timeout, ) @@ -630,10 +621,8 @@ def update_entry( "content": content, } ) - response = requests.put( + response = self._session.put( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -655,10 +644,8 @@ def update_entries(self, entry_ids: List[int], status: str) -> bool: """ endpoint = self._get_endpoint("/entries") data = {"entry_ids": entry_ids, "status": status} - response = requests.put( + response = self._session.put( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -678,8 +665,8 @@ def fetch_entry_content(self, entry_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/entries/{entry_id}/fetch-content") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -697,8 +684,8 @@ def toggle_bookmark(self, entry_id: int) -> bool: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/entries/{entry_id}/bookmark") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code >= 400: self._handle_error_response(response) @@ -716,8 +703,8 @@ def save_entry(self, entry_id: int) -> bool: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/entries/{entry_id}/save") - response = requests.post( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.post( + endpoint, timeout=self._timeout ) if response.status_code != 202: self._handle_error_response(response) @@ -735,8 +722,8 @@ def get_enclosure(self, enclosure_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/enclosures/{enclosure_id}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -758,10 +745,8 @@ def update_enclosure( """ endpoint = self._get_endpoint(f"/enclosures/{enclosure_id}") data = self._get_modification_params(media_progression=media_progression) - response = requests.put( + response = self._session.put( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -779,8 +764,8 @@ def get_categories(self) -> List[dict]: ClientError: If the request fails. """ endpoint = self._get_endpoint("/categories") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -799,8 +784,8 @@ def get_category_entry(self, category_id: int, entry_id: int) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/categories/{category_id}/entries/{entry_id}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -819,10 +804,8 @@ def get_category_entries(self, category_id: int, **kwargs) -> dict: """ endpoint = self._get_endpoint(f"/categories/{category_id}/entries") params = self._get_params(**kwargs) - response = requests.get( + response = self._session.get( endpoint, - headers=self._headers, - auth=self._auth, params=params, timeout=self._timeout, ) @@ -843,10 +826,8 @@ def create_category(self, title: str) -> dict: """ endpoint = self._get_endpoint("/categories") data = {"title": title} - response = requests.post( + response = self._session.post( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -868,10 +849,8 @@ def update_category(self, category_id: int, title: str) -> dict: """ endpoint = self._get_endpoint(f"/categories/{category_id}") data = {"id": category_id, "title": title} - response = requests.put( + response = self._session.put( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -889,8 +868,8 @@ def delete_category(self, category_id: int) -> None: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/categories/{category_id}") - response = requests.delete( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.delete( + endpoint, timeout=self._timeout ) if response.status_code != 204: self._handle_error_response(response) @@ -905,8 +884,8 @@ def mark_category_entries_as_read(self, category_id: int) -> None: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/categories/{category_id}/mark-all-as-read") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code != 204: self._handle_error_response(response) @@ -921,8 +900,8 @@ def get_users(self) -> List[dict]: ClientError: If the request fails. """ endpoint = self._get_endpoint("/users") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -956,8 +935,8 @@ def get_user_by_username(self, username: str) -> dict: def _get_user(self, user_id_or_username: Union[str, int]) -> dict: endpoint = self._get_endpoint(f"/users/{user_id_or_username}") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -978,10 +957,8 @@ def create_user(self, username: str, password: str, is_admin: bool = False) -> d """ endpoint = self._get_endpoint("/users") data = {"username": username, "password": password, "is_admin": is_admin} - response = requests.post( + response = self._session.post( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -1002,10 +979,8 @@ def update_user(self, user_id: int, **kwargs) -> dict: """ endpoint = self._get_endpoint(f"/users/{user_id}") data = self._get_modification_params(**kwargs) - response = requests.put( + response = self._session.put( endpoint, - headers=self._headers, - auth=self._auth, data=json.dumps(data), timeout=self._timeout, ) @@ -1023,8 +998,8 @@ def delete_user(self, user_id: int) -> None: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/users/{user_id}") - response = requests.delete( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.delete( + endpoint, timeout=self._timeout ) if response.status_code != 204: self._handle_error_response(response) @@ -1039,8 +1014,8 @@ def mark_user_entries_as_read(self, user_id: int) -> None: ClientError: If the request fails. """ endpoint = self._get_endpoint(f"/users/{user_id}/mark-all-as-read") - response = requests.put( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.put( + endpoint, timeout=self._timeout ) if response.status_code != 204: self._handle_error_response(response) @@ -1055,8 +1030,8 @@ def get_feed_counters(self) -> dict: ClientError: If the request fails. """ endpoint = self._get_endpoint("/feeds/counters") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json() @@ -1072,8 +1047,8 @@ def get_integrations_status(self) -> bool: ClientError: If the request fails. """ endpoint = self._get_endpoint("/integrations/status") - response = requests.get( - endpoint, headers=self._headers, auth=self._auth, timeout=self._timeout + response = self._session.get( + endpoint, timeout=self._timeout ) if response.status_code == 200: return response.json()["has_integrations"] diff --git a/tests/test_client.py b/tests/test_client.py index 8e73ec3..22dec3a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -34,6 +34,7 @@ ResourceNotFound, ServerError, ) +import requests from requests.exceptions import Timeout @@ -63,7 +64,7 @@ def test_get_error_with_bad_response(self): self.assertEqual(error.get_error_reason(), "status_code=404") def test_base_url_with_trailing_slash(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [ {"url": "http://example.org/feed", "title": "Example", "type": "RSS"} ] @@ -72,46 +73,41 @@ def test_base_url_with_trailing_slash(self): response.status_code = 200 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost/", "username", "password") + client = miniflux.Client("http://localhost/", "username", "password", session=session) result = client.discover("http://example.org/") - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/discover", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30.0, ) - + self.assertEqual(session.auth, ("username", "password")) self.assertEqual(result, expected_result) def test_flush_history(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 202 - requests.delete.return_value = response + session.delete = mock.Mock() + session.delete.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) result = client.flush_history() - requests.delete.assert_called_once_with( + session.delete.assert_called_once_with( "http://localhost/v1/flush-history", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30.0, ) - + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') self.assertTrue(result) def test_get_version(self): - requests = _get_request_mock() + session = requests.Session() expected_result = { "version": "dev", "commit": "HEAD", @@ -126,60 +122,56 @@ def test_get_version(self): response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) result = client.get_version() - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/version", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30.0, ) - + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') self.assertEqual(result, expected_result) def test_get_me(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "username": "foobar"} response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.me() - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/me", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) self.assertEqual(result, expected_result) def test_get_me_with_server_error(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 500 - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(ClientError): client.me() def test_discover(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [ {"url": "http://example.org/feed", "title": "Example", "type": "RSS"} ] @@ -188,20 +180,19 @@ def test_discover(self): response.status_code = 200 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.discover("http://example.org/") - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/discover", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("url"), "http://example.org/") @@ -210,7 +201,7 @@ def test_discover(self): self.assertEqual(result, expected_result) def test_discover_with_credentials(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [ {"url": "http://example.org/feed", "title": "Example", "type": "RSS"} ] @@ -219,9 +210,10 @@ def test_discover_with_credentials(self): response.status_code = 200 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.discover( "http://example.org/", username="foobar", @@ -229,15 +221,13 @@ def test_discover_with_credentials(self): user_agent="Bot", ) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/discover", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("url"), "http://example.org/") @@ -247,109 +237,106 @@ def test_discover_with_credentials(self): self.assertEqual(result, expected_result) def test_discover_with_server_error(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"error_message": "some error"} response = mock.Mock() response.status_code = 500 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(ClientError): client.discover("http://example.org/") def test_export(self): - requests = _get_request_mock() + session = requests.Session() expected_result = "OPML feed" response = mock.Mock() response.status_code = 200 response.text = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.export() - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/export", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) self.assertEqual(result, expected_result) def test_import(self): - requests = _get_request_mock() + session = requests.Session() input_data = "my opml data" response = mock.Mock() response.status_code = 201 - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) client.import_feeds(input_data) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/import", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, data=input_data, - auth=("username", "password"), timeout=30, ) def test_import_failure(self): - requests = _get_request_mock() + session = requests.Session() input_data = "my opml data" response = mock.Mock() response.status_code = 500 response.json.return_value = {"error_message": "random error"} - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(ClientError): client.import_feeds(input_data) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/import", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, data=input_data, - auth=("username", "password"), timeout=30, ) def test_get_feed(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "title": "Example"} response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_feed(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/feeds/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) self.assertEqual(result, expected_result) def test_get_feed_icon(self): - requests = _get_request_mock() + session = requests.Session() expected_result = { "id": 11, "mime_type": "image/x-icon", @@ -360,22 +347,21 @@ def test_get_feed_icon(self): response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_icon_by_feed_id(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/feeds/123/icon", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30.0, ) self.assertEqual(result, expected_result) def test_get_icon(self): - requests = _get_request_mock() + session = requests.Session() expected_result = { "id": 11, "mime_type": "image/x-icon", @@ -386,42 +372,40 @@ def test_get_icon(self): response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_icon(11) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/icons/11", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30.0, ) self.assertEqual(result, expected_result) def test_create_feed(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"feed_id": 42} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.create_feed("http://example.org/feed", 123) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/feeds", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("feed_url"), "http://example.org/feed") @@ -432,27 +416,26 @@ def test_create_feed(self): self.assertEqual(result, expected_result["feed_id"]) def test_create_feed_with_no_category(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"feed_id": 42} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.create_feed("http://example.org/feed") - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/feeds", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30.0, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("feed_url"), "http://example.org/feed") @@ -463,29 +446,28 @@ def test_create_feed_with_no_category(self): self.assertEqual(result, expected_result["feed_id"]) def test_create_feed_with_credentials(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"feed_id": 42} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.create_feed( "http://example.org/feed", 123, username="foobar", password="secret" ) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/feeds", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("feed_url"), "http://example.org/feed") @@ -496,27 +478,26 @@ def test_create_feed_with_credentials(self): self.assertEqual(result, expected_result["feed_id"]) def test_create_feed_with_crawler_enabled(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"feed_id": 42} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.create_feed("http://example.org/feed", 123, crawler=True) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/feeds", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("feed_url"), "http://example.org/feed") @@ -527,29 +508,28 @@ def test_create_feed_with_crawler_enabled(self): self.assertEqual(result, expected_result["feed_id"]) def test_create_feed_with_custom_user_agent_and_crawler_disabled(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"feed_id": 42} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.create_feed( "http://example.org/feed", 123, crawler=False, user_agent="GoogleBot" ) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/feeds", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.post.call_args + _, kwargs = session.post.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("feed_url"), "http://example.org/feed") @@ -561,27 +541,26 @@ def test_create_feed_with_custom_user_agent_and_crawler_disabled(self): self.assertEqual(result, expected_result["feed_id"]) def test_update_feed(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "crawler": True, "username": "test"} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.update_feed(123, crawler=True, username="test") - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/feeds/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.put.call_args + _, kwargs = session.put.call_args payload = json.loads(kwargs.get("data")) self.assertNotIn("feed_url", payload) @@ -591,110 +570,105 @@ def test_update_feed(self): self.assertEqual(result, expected_result) def test_refresh_all_feeds(self): - requests = _get_request_mock() + session = requests.Session() expected_result = True response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.refresh_all_feeds() - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/feeds/refresh", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_refresh_feed(self): - requests = _get_request_mock() + session = requests.Session() expected_result = True response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.refresh_feed(123) - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/feeds/123/refresh", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_refresh_category(self): - requests = _get_request_mock() + session = requests.Session() expected_result = True response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.refresh_category(123) - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/categories/123/refresh", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_get_feed_entry(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {} response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_feed_entry(123, 456) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/feeds/123/entries/456", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_get_feed_entries(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_feed_entries(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/feeds/123/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params=None, timeout=30, ) @@ -702,22 +676,21 @@ def test_get_feed_entries(self): assert result == expected_result def test_get_feed_entries_with_direction_param(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_feed_entries(123, direction="asc") - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/feeds/123/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params={"direction": "asc"}, timeout=30, ) @@ -725,129 +698,117 @@ def test_get_feed_entries_with_direction_param(self): assert result == expected_result def test_mark_feed_as_read(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 204 - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) client.mark_feed_entries_as_read(123) - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/feeds/123/mark-all-as-read", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30, ) + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') def test_mark_category_entries_as_read(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 204 - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) client.mark_category_entries_as_read(123) - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/categories/123/mark-all-as-read", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30, ) + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') def test_mark_user_entries_as_read(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 204 - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) client.mark_user_entries_as_read(123) - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/users/123/mark-all-as-read", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30, ) + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') def test_get_entry(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_entry(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/entries/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_fetch_entry_content(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.fetch_entry_content(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/entries/123/fetch-content", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_get_entries(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_entries(status="unread", limit=10, offset=5) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params=mock.ANY, timeout=30, ) @@ -856,22 +817,21 @@ def test_get_entries(self): def test_get_entries_with_before_param(self): param_value = int(time.time()) - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_entries(before=param_value) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params={"before": param_value}, timeout=30, ) @@ -879,22 +839,21 @@ def test_get_entries_with_before_param(self): assert result == expected_result def test_get_entries_with_starred_param(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_entries(starred=True) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params={"starred": True}, timeout=30, ) @@ -902,22 +861,21 @@ def test_get_entries_with_starred_param(self): assert result == expected_result def test_get_entries_with_starred_param_at_false(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_entries(starred=False, after_entry_id=123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params={"after_entry_id": 123}, timeout=30, ) @@ -925,85 +883,83 @@ def test_get_entries_with_starred_param_at_false(self): assert result == expected_result def test_get_user_by_id(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_user_by_id(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/users/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_get_inexisting_user(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 404 response.json.return_value = {"error_message": "some error"} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(ResourceNotFound): client.get_user_by_id(123) def test_get_user_by_username(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_user_by_username("foobar") - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/users/foobar", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_update_user(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "theme": "Black", "language": "fr_FR"} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.update_user(123, theme="black", language="fr_FR") - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/users/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30, ) - _, kwargs = requests.put.call_args + _, kwargs = session.put.call_args payload = json.loads(kwargs.get("data")) self.assertNotIn("username", payload) @@ -1013,103 +969,94 @@ def test_update_user(self): self.assertEqual(result, expected_result) def test_timeout(self): - requests = _get_request_mock() - requests.get.side_effect = Timeout() + session = requests.Session() + session.get = mock.Mock() + session.get.side_effect = Timeout() - client = miniflux.Client("http://localhost", "username", "password", 1.0) + client = miniflux.Client("http://localhost", "username", "password", 1.0, session=session) with self.assertRaises(Timeout): client.export() - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/export", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=1.0, ) def test_api_key_auth(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 200 response.json.return_value = {} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) client.export() - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/export", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30.0, ) + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') def test_save_entry(self): - requests = _get_request_mock() + session = requests.Session() expected_result = True response = mock.Mock() response.status_code = 202 - requests.post.return_value = response + session.post = mock.Mock() + session.post.return_value = response - client = miniflux.Client("http://localhost", api_key="secret") + client = miniflux.Client("http://localhost", api_key="secret", session=session) result = client.save_entry(123) - requests.post.assert_called_once_with( + session.post.assert_called_once_with( "http://localhost/v1/entries/123/save", - headers={ - "User-Agent": miniflux.DEFAULT_USER_AGENT, - "X-Auth-Token": "secret", - }, - auth=None, timeout=30.0, ) - assert result == expected_result + self.assertEqual(session.headers.get('X-Auth-Token'), 'secret') + self.assertEqual(result, expected_result) def test_get_category_entry(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {} response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_category_entry(123, 456) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/categories/123/entries/456", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30, ) assert result == expected_result def test_get_category_entries(self): - requests = _get_request_mock() + session = requests.Session() expected_result = [] response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_category_entries(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/categories/123/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), params=None, timeout=30, ) @@ -1117,79 +1064,76 @@ def test_get_category_entries(self): assert result == expected_result def test_update_entry_title(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "title": "New title"} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.update_entry(entry_id=123, title="New title") - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/entries/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30.0, ) - _, kwargs = requests.put.call_args + _, kwargs = session.put.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("title"), "New title") self.assertEqual(result, expected_result) def test_update_entry_content(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "content": "New content"} response = mock.Mock() response.status_code = 201 response.json.return_value = expected_result - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.update_entry(entry_id=123, content="New content") - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/entries/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30.0, ) - _, kwargs = requests.put.call_args + _, kwargs = session.put.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("content"), "New content") self.assertEqual(result, expected_result) def test_update_entries_status(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 204 - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.update_entries(entry_ids=[123, 456], status="read") - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/entries", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30.0, ) - _, kwargs = requests.put.call_args + _, kwargs = session.put.call_args payload = json.loads(kwargs.get("data")) self.assertEqual(payload.get("entry_ids"), [123, 456]) @@ -1197,139 +1141,136 @@ def test_update_entries_status(self): self.assertTrue(result) def test_get_enclosure(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"id": 123, "mime_type": "audio/mpeg"} response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_enclosure(123) - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/enclosures/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30.0, ) self.assertEqual(result, expected_result) def test_update_enclosure(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 204 - requests.put.return_value = response + session.put = mock.Mock() + session.put.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) self.assertTrue(client.update_enclosure(123, media_progression=42)) - requests.put.assert_called_once_with( + session.put.assert_called_once_with( "http://localhost/v1/enclosures/123", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), data=mock.ANY, timeout=30.0, ) def test_get_integrations_status(self): - requests = _get_request_mock() + session = requests.Session() expected_result = {"has_integrations": True} response = mock.Mock() response.status_code = 200 response.json.return_value = expected_result - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) result = client.get_integrations_status() - requests.get.assert_called_once_with( + session.get.assert_called_once_with( "http://localhost/v1/integrations/status", - headers={"User-Agent": miniflux.DEFAULT_USER_AGENT}, - auth=("username", "password"), timeout=30.0, ) self.assertTrue(result) def test_not_found_response(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 404 response.json.return_value = {"error_message": "Not found"} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(ResourceNotFound): client.get_version() def test_unauthorized_response(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 401 response.json.return_value = {"error_message": "Unauthorized"} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(AccessUnauthorized): client.get_version() def test_forbidden_response(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 403 response.json.return_value = {"error_message": "Forbidden"} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(AccessForbidden): client.get_version() def test_bad_request_response(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 400 response.json.return_value = {"error_message": "Bad request"} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(BadRequest): client.get_version() def test_server_error_response(self): - requests = _get_request_mock() + session = requests.Session() response = mock.Mock() response.status_code = 500 response.json.return_value = {"error_message": "Server error"} - requests.get.return_value = response + session.get = mock.Mock() + session.get.return_value = response - client = miniflux.Client("http://localhost", "username", "password") + client = miniflux.Client("http://localhost", "username", "password", session=session) with self.assertRaises(ServerError): client.get_version() - - -def _get_request_mock(): - patcher = mock.patch("miniflux.requests") - return patcher.start() From 276e4a7ee641cb4e9e388fcea215ba586a7d719f Mon Sep 17 00:00:00 2001 From: Florian S Date: Sun, 26 Jan 2025 16:31:35 +0100 Subject: [PATCH 2/4] fix linter error --- miniflux.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/miniflux.py b/miniflux.py index e9ce552..4bbacb0 100644 --- a/miniflux.py +++ b/miniflux.py @@ -123,7 +123,7 @@ def __init__( self._session.headers.update({"User-Agent": user_agent}) if api_key: self._session.headers.update({"X-Auth-Token": api_key}) - if auth != None: + if auth is not None: self._session.auth = auth def _get_endpoint(self, path: str) -> str: From c5e71a77e3543522b47b6a09756db7999df86b36 Mon Sep 17 00:00:00 2001 From: Florian S Date: Sun, 26 Jan 2025 16:33:40 +0100 Subject: [PATCH 3/4] Make the miniflux client a context manager To handle the underlying session --- miniflux.py | 12 ++++++++++++ tests/test_client.py | 26 ++++++++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/miniflux.py b/miniflux.py index 4bbacb0..2e4cfa0 100644 --- a/miniflux.py +++ b/miniflux.py @@ -126,6 +126,12 @@ def __init__( if auth is not None: self._session.auth = auth + def __enter__(self): + return self + + def __exit__(self, *args): + self.close() + def _get_endpoint(self, path: str) -> str: if len(self._base_url) > 0 and self._base_url[-1:] == "/": self._base_url = self._base_url[:-1] @@ -1053,3 +1059,9 @@ def get_integrations_status(self) -> bool: if response.status_code == 200: return response.json()["has_integrations"] self._handle_error_response(response) + + def close(self) -> None: + """ + Close the underlying session + """ + self._session.close() diff --git a/tests/test_client.py b/tests/test_client.py index 22dec3a..be3ec30 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1274,3 +1274,29 @@ def test_server_error_response(self): with self.assertRaises(ServerError): client.get_version() + + def test_session_closed(self): + session = mock.Mock() + + client = miniflux.Client("http://localhost", "username", "password", session=session) + client.close() + + session.close.assert_called() + + def test_context_manager_exit_on_error(self): + response = mock.Mock() + response.status_code = 500 + response.json.return_value = {"error_message": "Server error"} + + session = mock.Mock() + session.get.return_value = response + + with ( + miniflux.Client("http://localhost", "username", "password", session=session) as client, + self.assertRaises(ServerError) + ): + client.get_version() + + session.close.assert_called() + + From 86514fb1fabce38c0c0214c394df784e85a6f9c8 Mon Sep 17 00:00:00 2001 From: Florian S Date: Sun, 26 Jan 2025 20:00:46 +0100 Subject: [PATCH 4/4] fix test for python 3.6 --- tests/test_client.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/test_client.py b/tests/test_client.py index be3ec30..7cd17b1 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1291,12 +1291,12 @@ def test_context_manager_exit_on_error(self): session = mock.Mock() session.get.return_value = response - with ( - miniflux.Client("http://localhost", "username", "password", session=session) as client, - self.assertRaises(ServerError) - ): - client.get_version() - + with miniflux.Client( + "http://localhost", "username", "password", session=session + ) as client: + with self.assertRaises(ServerError): + client.get_version() + session.close.assert_called()