add last_refresh_timestamp

Author: rsgowman

firebase_admin/_user_mgt.py

@@ -19,6 +19,7 @@
 from urllib import parse
 
 import requests
+import iso8601
 
 from firebase_admin import _auth_utils
 from firebase_admin import _identifier
@@ -42,11 +43,14 @@ def __init__(self, description):
 class UserMetadata:
     """Contains additional metadata associated with a user account."""
 
-    def __init__(self, creation_timestamp=None, last_sign_in_timestamp=None):
+    def __init__(self, creation_timestamp=None, last_sign_in_timestamp=None,
+                 last_refresh_timestamp=None):
         self._creation_timestamp = _auth_utils.validate_timestamp(
             creation_timestamp, 'creation_timestamp')
         self._last_sign_in_timestamp = _auth_utils.validate_timestamp(
             last_sign_in_timestamp, 'last_sign_in_timestamp')
+        self._last_refresh_timestamp = _auth_utils.validate_timestamp(
+            last_refresh_timestamp, 'last_refresh_timestamp')
 
     @property
     def creation_timestamp(self):
@@ -66,6 +70,16 @@ def last_sign_in_timestamp(self):
         """
         return self._last_sign_in_timestamp
 
+    @property
+    def last_refresh_timestamp(self):
+        """The time at which the user was last active (ID token refreshed).
+
+        Returns:
+          integer: Milliseconds since epoch timestamp, or None if the user was
+            never active.
+        """
+        return self._last_refresh_timestamp
+
 
 class UserInfo:
     """A collection of standard profile information for a user.
@@ -217,7 +231,12 @@ def _int_or_none(key):
             if key in self._data:
                 return int(self._data[key])
             return None
-        return UserMetadata(_int_or_none('createdAt'), _int_or_none('lastLoginAt'))
+        last_refresh_at_millis = None
+        last_refresh_at_iso8601 = self._data.get('lastRefreshAt', None)
+        if last_refresh_at_iso8601 is not None:
+            last_refresh_at_millis = iso8601.parse_date(last_refresh_at_iso8601).timestamp() * 1000
+        return UserMetadata(
+            _int_or_none('createdAt'), _int_or_none('lastLoginAt'), last_refresh_at_millis)
 
     @property
     def provider_data(self):

integration/test_auth.py

@@ -46,7 +46,7 @@ def _sign_in(custom_token, api_key):
     return resp.json().get('idToken')
 
 def _sign_in_with_password(email, password, api_key):
-    body = {'email': email, 'password': password}
+    body = {'email': email, 'password': password, 'returnSecureToken': True}
     params = {'key' : api_key}
     resp = requests.request('post', _verify_password_url, params=params, json=body)
     resp.raise_for_status()
@@ -162,7 +162,7 @@ def new_user():
     auth.delete_user(user.uid)
 
 @pytest.fixture
-def new_user_with_params():
+def new_user_with_params() -> auth.UserRecord:
     random_id, email = _random_id()
     phone = _random_phone()
     user = auth.create_user(
@@ -281,6 +281,25 @@ def test_de_dups_duplicate_users(self):
         users = map(self._map_user_record_to_uid_email_phones, users)
         assert list(users) == [self.test_user1]
 
+def test_last_refresh_timestamp(new_user_with_params: auth.UserRecord, api_key):
+    # new users should not have a last_refresh_timestamp set
+    assert new_user_with_params.user_metadata.last_refresh_timestamp is None
+
+    # login to cause the last_refresh_timestamp to be set
+    _sign_in_with_password(new_user_with_params.email, 'secret', api_key)
+    new_user_with_params = auth.get_user(new_user_with_params.uid)
+
+    # Ensure the last refresh time occurred at approximately 'now'. (With a
+    # tolerance of up to 1 minute; we ideally want to ensure that any timezone
+    # considerations are handled properly, so as long as we're within an hour,
+    # we're in good shape.)
+    millis_per_second = 1000
+    seconds_per_minute = 60
+    millis_per_minute = millis_per_second * seconds_per_minute
+
+    last_refresh_timestamp = new_user_with_params.user_metadata.last_refresh_timestamp
+    assert last_refresh_timestamp == pytest.approx(
+        time.time()*millis_per_second, 1*millis_per_minute)
 
 def test_list_users(new_user_list):
     err_msg_template = (

requirements.txt

@@ -8,3 +8,4 @@ google-api-core[grpc] >= 1.14.0, < 2.0.0dev; platform.python_implementation != '
 google-api-python-client >= 1.7.8
 google-cloud-firestore >= 1.4.0; platform.python_implementation != 'PyPy'
 google-cloud-storage >= 1.18.0
+iso8601 >= 0.1.12