auth reset command (#11)

* reset command

* WIP

* formatting and test fix

* cleanup

* remove debug code

* update comments

---------

Co-authored-by: Carl A. Adams <[email protected]>

Author: carl-alexander-adams

src/planet_auth/constants.py

@@ -19,5 +19,6 @@
 TOKEN_FILE_PLAIN = "token.json"
 TOKEN_FILE_SOPS = "token.sops.json"
 USER_CONFIG_FILE = ".planet.json"
+PROFILE_DIR = ".planet"
 X_PLANET_APP = f"planet-auth-library-{importlib.metadata.version('planet-auth')}"
 X_PLANET_APP_HEADER = "X-Planet-App"

src/planet_auth/storage_utils.py

@@ -63,6 +63,12 @@ def obj_exists(self, key: ObjectStorageProvider_KeyType) -> bool:
         Check whether a given object exists in storage.
         """
 
+    @abstractmethod
+    def obj_rename(self, src: ObjectStorageProvider_KeyType, dst: ObjectStorageProvider_KeyType) -> None:
+        """
+        Rename/Move an object from the source path to the destination path.
+        """
+
     @staticmethod
     def _default_storage_provider():
         # Always create JIT to better handle cases where runtime changes to
@@ -175,6 +181,11 @@ def obj_exists(self, key: ObjectStorageProvider_KeyType) -> bool:
         obj_filepath = self._obj_filepath(key)
         return obj_filepath.exists()
 
+    def obj_rename(self, src: ObjectStorageProvider_KeyType, dst: ObjectStorageProvider_KeyType) -> None:
+        src_filepath = self._obj_filepath(src)
+        dst_filepath = self._obj_filepath(dst)
+        src_filepath.rename(dst_filepath)
+
 
 class FileBackedJsonObjectException(AuthException):
     """

src/planet_auth_utils/__init__.py

@@ -20,7 +20,12 @@
 
 """
 
-from .commands.cli.main import cmd_plauth_embedded, cmd_plauth_login
+from .commands.cli.main import (
+    cmd_plauth_embedded,
+    cmd_plauth_login,
+    cmd_plauth_reset,
+    cmd_plauth_version,
+)
 from .commands.cli.planet_legacy_auth_cmd import (
     cmd_pllegacy,
     cmd_pllegacy_login,
@@ -91,6 +96,8 @@
 __all__ = [
     "cmd_plauth_embedded",
     "cmd_plauth_login",
+    "cmd_plauth_reset",
+    "cmd_plauth_version",
     "cmd_jwt",
     "cmd_jwt_decode",
     "cmd_jwt_validate_oauth",

src/planet_auth_utils/commands/cli/main.py

@@ -16,10 +16,13 @@
 import logging
 import importlib.metadata
 import sys
+import time
 
-from planet_auth import Auth, AuthException, setStructuredLogging
+from planet_auth import Auth, AuthException, setStructuredLogging, ObjectStorageProvider, ObjectStorageProvider_KeyType
+from planet_auth.constants import USER_CONFIG_FILE
 
 from planet_auth_utils.plauth_factory import PlanetAuthFactory
+from planet_auth_utils.profile import Profile
 
 from .options import (
     opt_organization,
@@ -119,6 +122,29 @@ def _pkg_display_version(pkg_name):
     print(f"planet : {_pkg_display_version('planet')}")
 
 
+@cmd_plauth.command("reset")
+def cmd_plauth_reset():
+    """
+    Reset saved auth state.
+
+    Old auth state is not deleted. It is moved aside and preserved.
+    """
+    save_tag = time.strftime("%Y-%m-%d-%H%M%S")
+
+    # The CLI only supports the default storage provider right now.
+    storage_provider = ObjectStorageProvider._default_storage_provider()
+
+    user_conf_objkey = ObjectStorageProvider_KeyType(USER_CONFIG_FILE)
+    if storage_provider.obj_exists(user_conf_objkey):
+        user_conf_objkey_offname = ObjectStorageProvider_KeyType(USER_CONFIG_FILE + f"-{save_tag}")
+        storage_provider.obj_rename(user_conf_objkey, user_conf_objkey_offname)
+
+    profile_dir_objkey = ObjectStorageProvider_KeyType(Profile.profile_root())
+    if storage_provider.obj_exists(profile_dir_objkey):
+        profile_dir_objkey_offname = ObjectStorageProvider_KeyType(Profile.profile_root().name + f"-{save_tag}")
+        storage_provider.obj_rename(profile_dir_objkey, profile_dir_objkey_offname)
+
+
 @cmd_plauth.command("login")
 @opt_open_browser
 @opt_show_qr_code

src/planet_auth_utils/commands/cli/options.py

@@ -15,6 +15,7 @@
 import click
 import pathlib
 
+from planet_auth import constants
 from planet_auth_utils.constants import EnvironmentVariables
 
 
@@ -76,7 +77,7 @@ def opt_profile(function):
         type=str,
         envvar=EnvironmentVariables.AUTH_PROFILE,
         help="Select the client profile to use.  User created profiles are "
-        " defined by creating a subdirectory ~/.planet/.  Additionally, a number of"
+        f" defined by creating a subdirectory ~/{constants.PROFILE_DIR}/.  Additionally, a number of"
         ' built-in profiles are understood.  See the "profile list" command'
         " for defined profiles.  The auth profile controls how the software"
         " interacts with authentication services, as well as how it"

src/planet_auth_utils/profile.py

@@ -21,6 +21,7 @@
 from planet_auth.constants import (
     AUTH_CONFIG_FILE_SOPS,
     AUTH_CONFIG_FILE_PLAIN,
+    PROFILE_DIR,
 )
 from planet_auth.storage_utils import ObjectStorageProvider
 
@@ -43,13 +44,13 @@ def profile_root() -> pathlib.Path:
         """
         Root storage directory used for profile data.
         """
-        # return pathlib.Path.home().joinpath(".planet")
+        # return pathlib.Path.home().joinpath(PROFILE_DIR)
         # We used to assume file storage.  We now support pluggable storage implementation.
         # We let the storage provider determine the real path (if there is one) and in this
         # Profile class we now (mostly) deal in the abstract idea of the path as an object
         # identifier that may or may not be file system based, depending on the storage
         # provider in use.
-        return pathlib.Path(".planet")
+        return pathlib.Path(PROFILE_DIR)
 
     @staticmethod
     def get_profile_dir_path(profile: str) -> pathlib.Path:

tests/test_planet_auth/unit/auth/util.py

@@ -454,3 +454,10 @@ def save_obj(self, key: ObjectStorageProvider_KeyType, data: dict) -> None:
 
     def obj_exists(self, key: ObjectStorageProvider_KeyType) -> bool:
         return key in self._mock_storage
+
+    def obj_rename(self, src: ObjectStorageProvider_KeyType, dst: ObjectStorageProvider_KeyType) -> None:
+        if self._mock_storage[src] is not None:
+            self._mock_storage[dst] = self._mock_storage[src]
+            del self._mock_storage[src]
+        else:
+            del self._mock_storage[dst]