MFAコードを使用したログインができるようにした (#613)

* MFAコードを使用したログインができるようにした

* 自動生成しない旨をコメントで残す

* リファクタリング

* refresh_token関数内でlogin関数を実行する

* login関数に`@my_backoff`は不要なので削除

* API v2もrefresh_tokenを利用する

* 不具合修正

* format

---------

Co-authored-by: yuji38kwmt <yuji38kwmt@yahoo.co.jp>

annofabapi/api.py

@@ -512,7 +512,7 @@ def _request_wrapper(
 
         # Unauthorized Errorならば、ログイン後に再度実行する
         if response.status_code == requests.codes.unauthorized:
-            self.login()
+            self.refresh_token()
             return self._request_wrapper(
                 http_method,
                 url_path,
@@ -620,12 +620,13 @@ def _request_get_with_cookie(self, project_id: str, url: str) -> requests.Respon
     #########################################
     # Public Method : Login
     #########################################
-    @my_backoff
-    def login(self) -> Tuple[Dict[str, Any], requests.Response]:
+    def login(self, mfa_code: Optional[str] = None) -> None:
         """
-        ログイン
-
+        ログインして、トークンをインスタンスに保持します。
+        MFAが有効化されている場合は、loginRespondToAuthChallenge APIを実行してトークンを取得します。
 
+        Args:
+            mfa_code: ``loginRespondToAuthChallenge``のレスポンスから取得したMFAコード。この引数はexperimentalです。将来削除される可能性があります。
 
         Returns:
             Tuple[Token, requests.Response]
@@ -635,19 +636,28 @@ def login(self) -> Tuple[Dict[str, Any], requests.Response]:
 
         url = f"{self.url_prefix}/login"
 
-        response = self._execute_http_request("post", url, json=login_info)
-        json_obj = response.json()
+        login_response = self._execute_http_request("post", url, json=login_info)
+        json_obj = login_response.json()
         if "token" not in json_obj:
-            raise MfaEnabledUserExecutionError(self.login_user_id)
-        self.token_dict = json_obj["token"]
-
+            # `login` APIのレスポンスのスキーマがLoginNeedChallengeResponseのとき
+            if mfa_code is None:
+                raise MfaEnabledUserExecutionError(self.login_user_id)
+
+            mfa_param = {"user_id": self.login_user_id, "mfa_code": mfa_code, "session": json_obj["session"]}
+            mfa_url = f"{self.url_prefix}/login-respond-to-auth-challenge"
+            mfa_response = self._execute_http_request("post", mfa_url, json=mfa_param)
+            mfa_json_obj = mfa_response.json()
+            token_dict = mfa_json_obj["token"]
+        else:
+            # `login` APIのレスポンスのスキーマがloginRespondToAuthChallengeのとき
+            token_dict = json_obj["token"]
+        self.token_dict = token_dict
         logger.debug("Logged in successfully. user_id = %s", self.login_user_id)
-        return json_obj, response
 
-    def logout(self) -> Tuple[Dict[str, Any], requests.Response]:
+    def logout(self) -> None:
         """
-        ログアウト
-        ログインしていないときはNoneを返す。
+        ログアウトします。
+        ログアウト後は、インスタンス変数 ``token_dict`` をNoneにします。
 
 
 
@@ -662,32 +672,32 @@ def logout(self) -> Tuple[Dict[str, Any], requests.Response]:
             raise NotLoggedInError
 
         request_body = self.token_dict
-        content, response = self._request_wrapper("POST", "/logout", request_body=request_body)
+        url = f"{self.url_prefix}/logout"
+        self._execute_http_request("POST", url, json=request_body)
         self.token_dict = None
-        return content, response
 
-    def refresh_token(self) -> Tuple[Dict[str, Any], requests.Response]:
+    def refresh_token(self) -> None:
         """
-        トークン リフレッシュ
-        ログインしていないときはNoneを返す。
-
-
-
-        Returns:
-            Tuple[Token, requests.Response]
-
-        Raises:
-            NotLoggedInError: ログインしてない状態で関数を呼び出したときのエラー
+        トークンを再発行して、新しいトークン情報をインスタンスに保持します。
+        ログインしていない場合やリフレッシュトークンの有効期限が切れている場合は、login APIを実行して新しいトークン情報をインスタンスに保持します。
 
         """
 
         if self.token_dict is None:
-            raise NotLoggedInError
+            # 一度もログインしていないときは、login APIを実行して、トークン情報をインスタンスに保持する（login関数内でインスタンスに保持している）
+            self.login()
+            return
 
         request_body = {"refresh_token": self.token_dict["refresh_token"]}
-        content, response = self._request_wrapper("POST", "/refresh-token", request_body=request_body)
-        self.token_dict = content
-        return content, response
+        url = f"{self.url_prefix}/refresh-token"
+        response = self._execute_http_request("POST", url, json=request_body)
+
+        # Unauthorized Errorならば、login APIを実行して、取得したトークン情報をインスタンスに保持する
+        if response.status_code == requests.codes.unauthorized:
+            self.login()
+            return
+
+        self.token_dict = response.json()
 
     #########################################
     # Public Method : Other

annofabapi/api2.py

@@ -78,7 +78,7 @@ def _request_wrapper(
 
             # Unauthorized Errorならば、ログイン後に再度実行する
             if response.status_code == requests.codes.unauthorized:
-                self.api.login()
+                self.api.refresh_token()
                 return self._request_wrapper(
                     http_method,
                     url_path,

annofabapi/models.py

@@ -6337,6 +6337,9 @@ class WebhookStatus(Enum):
 """
 
 
+# `@deprecated_class`を指定した理由：
+# InspectionStatusは非推奨のgetInspections APIとbatchUpdateComments APIからしか参照されていない。
+# したがってInspectionStatusも非推奨にしている。
 @deprecated_class(deprecated_date="2022-08-23")
 class InspectionStatus(Enum):
     """

annofabapi/wrapper.py

@@ -2366,6 +2366,10 @@ def can_execute_job(self, project_id: str, job_type: ProjectJobType) -> bool:
             project_id: プロジェクトID
             job_type: ジョブ種別
 
+
+        Notes:
+            MFAが有効でログインしていない場合、login APIでエラーが発生します。事前にログインしておく必要があります。
+
         Returns:
             ジョブが実行できる状態か否か
         """
@@ -2401,6 +2405,9 @@ def wait_until_job_is_executable(
         Returns:
             指定した時間（アクセス頻度と回数）待った後、ジョブが実行可能な状態かどうか。進行中のジョブが存在する場合は、ジョブが実行不可能。
 
+        Notes:
+            MFAが有効でログインしていない場合、login APIでエラーが発生します。事前にログインしておく必要があります。
+
         """
 
         job_type_list = _JOB_CONCURRENCY_LIMIT[job_type]

generate/.openapi-generator-ignore_v1

@@ -1,2 +1,10 @@
+# 自動生成しない理由：
+# ログイン関係のAPIは直接利用すると使いにくいので、APIとは異なる引数にしている（たとえばlogin関数は引数でuser_id,passwordを受け取るのではなく、インスタンスにあるuser_id, passwordを参照するなど）
+# したがって、自動生成しない
 out/openapi_client/api/af_login_api.py
+
+# 自動生成しない理由：
+# InspectionStatusは非推奨のgetInspections APIとbatchUpdateComments APIからしか参照されていない。
+# したがってInspectionStatusも非推奨にしている。
+# `models.py`に非推奨であるデコレータを付与しているので、自動で生成されるとデコレータの部分が消えてしまう。したがって、自動生成しない。
 out/openapi_client/models/inspection_status.py

generate/partial-footer/models_partial_footer.py

@@ -1,3 +1,6 @@
+# `@deprecated_class`を指定した理由：
+# InspectionStatusは非推奨のgetInspections APIとbatchUpdateComments APIからしか参照されていない。
+# したがってInspectionStatusも非推奨にしている。
 @deprecated_class(deprecated_date="2022-08-23")
 class InspectionStatus(Enum):
     """

tests/test_api.py

@@ -27,7 +27,6 @@
 from annofabapi.dataclass.project_member import ProjectMember
 from annofabapi.dataclass.supplementary import SupplementaryData
 from annofabapi.dataclass.task import Task, TaskHistory
-from annofabapi.exceptions import NotLoggedInError
 from annofabapi.models import GraphType, ProjectJobType
 from annofabapi.wrapper import TaskFrameKey
 from tests.utils_for_test import WrapperForTest, create_csv_for_task
@@ -300,19 +299,10 @@ def test_scenario(self):
 
 class TestLogin:
     def test_login(self):
-        assert api.login()[0]["token"].keys() >= {"id_token", "access_token", "refresh_token"}
-
-        assert api.refresh_token()[0].keys() >= {"id_token", "access_token", "refresh_token"}
-
-        assert type(api.logout()[0]) == dict
-
-        # ログアウト状態では、refresh_tokenメソッドはExceptionをスローする
-        with pytest.raises(NotLoggedInError):
-            api.refresh_token()
-
-        # ログアウト状態では、logoutメソッドはNoneを返す
-        with pytest.raises(NotLoggedInError):
-            api.logout()
+        # Exceptionをスローしないことの確認
+        api.login()
+        api.refresh_token()
+        api.logout()
 
 
 class TestMy: