Merge pull request #166 from python-astrodynamics/feature/session-state

Fix closing after a failed logout and changing base_url

Author: RazerM

newsfragments/166.fixed.2.rst

@@ -0,0 +1 @@
+Setting ``base_url`` now resets the authentication state and the cached predicates, which are specific to the previous host.

newsfragments/166.fixed.rst

@@ -0,0 +1 @@
+:meth:`SpaceTrackClient.close() <spacetrack.base.SpaceTrackClient.close>` and :meth:`AsyncSpaceTrackClient.close() <spacetrack.aio.AsyncSpaceTrackClient.close>` now close the underlying HTTPX client even if logging out fails.

src/spacetrack/aio.py

@@ -302,8 +302,10 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
     async def close(self):
         """Log out of Space-Track (if necessary) and close any open connections."""
         self._finalizer.detach()
-        await self.logout()
-        await self.client.aclose()
+        try:
+            await self.logout()
+        finally:
+            await self.client.aclose()
 
 
 async def _iter_lines_generator(response):

src/spacetrack/base.py

@@ -400,6 +400,9 @@ def base_url(self):
     @base_url.setter
     def base_url(self, url):
         self.client.base_url = url
+        # The session cookie and predicate metadata are host-specific.
+        self._authenticated = False
+        self._predicates = dict()
 
     def _handle_event(self, event):
         if isinstance(event, NormalRequest):
@@ -1089,8 +1092,10 @@ def _cleanup(cls, warn_message):
     def close(self):
         """Log out of Space-Track (if necessary) and close any open connections."""
         self._finalizer.detach()
-        self.logout()
-        self.client.close()
+        try:
+            self.logout()
+        finally:
+            self.client.close()
 
     def __repr__(self):
         r = ReprHelper(self)

tests/test_spacetrack.py

@@ -504,6 +504,22 @@ def test_base_url(httpx2_mock):
     assert len(httpx2_mock.get_requests(method="POST", url=login_url)) == 1
 
 
+def test_base_url_change_resets_session_state(client, httpx2_mock, mock_gp_predicates):
+    httpx2_mock.add_response(method="POST", url=api_url("ajaxauth/login"), json="")
+    httpx2_mock.add_response(
+        method="GET", url=api_url("basicspacedata/query/class/gp"), json={"a": 1}
+    )
+
+    assert client.gp() == {"a": 1}
+    assert client._authenticated
+    assert client._predicates
+
+    client.base_url = "https://testing.space-track.org"
+
+    assert not client._authenticated
+    assert not client._predicates
+
+
 def test_raise_for_status(httpx2_mock):
     httpx2_mock.add_response(
         method="GET",
@@ -781,6 +797,24 @@ def test_unknown_event(client):
         client._handle_event(object())
 
 
+def test_close_when_logout_fails(httpx2_mock):
+    httpx2_mock.add_response(method="POST", url=api_url("ajaxauth/login"), json="")
+    httpx2_mock.add_response(
+        method="GET",
+        url=api_url("ajaxauth/logout"),
+        status_code=500,
+        json={"error": "oops"},
+    )
+
+    client = SpaceTrackClient("identity", "password")
+    client.authenticate()
+
+    with pytest.raises(httpx2.HTTPStatusError):
+        client.close()
+
+    assert client.client.is_closed
+
+
 def test_implicit_cleanup_warning():
     with pytest.warns(ResourceWarning, match="without being closed explicitly"):
         SpaceTrackClient("identity", "password")