Feat/operator logins

cuenca/__init__.py

@@ -26,6 +26,8 @@
     'LoginToken',
     'Otp',
     'OperatorLogin',
+    'OperatorOtp',
+    'OperatorToken',
     'PasswordReset',
     'Platform',
     'Questionnaires',
@@ -82,6 +84,8 @@
     LimitedWallet,
     LoginToken,
     OperatorLogin,
+    OperatorOtp,
+    OperatorToken,
     Otp,
     PasswordReset,
     PhoneVerificationAssociations,

cuenca/resources/__init__.py

@@ -26,6 +26,8 @@
     'LoginToken',
     'Otp',
     'OperatorLogin',
+    'OperatorOtp',
+    'OperatorToken',
     'PasswordReset',
     'Platform',
     'PhoneVerificationAssociation',
@@ -76,6 +78,8 @@
 from .limited_wallets import LimitedWallet
 from .login_tokens import LoginToken
 from .operator_logins import OperatorLogin
+from .operator_otps import OperatorOtp
+from .operator_tokens import OperatorToken
 from .otps import Otp
 from .password_resets import PasswordReset
 from .phone_verification_associations import PhoneVerificationAssociations
@@ -127,6 +131,8 @@
     LimitedWallet,
     LoginToken,
     OperatorLogin,
+    OperatorOtp,
+    OperatorToken,
     PasswordReset,
     Questionnaires,
     Saving,

cuenca/resources/operator_logins.py

@@ -1,23 +1,47 @@
-from typing import Annotated, ClassVar
+import datetime as dt
+from typing import Annotated, ClassVar, Optional
 
 from cuenca_validations.types import LogConfig, OperatorRole
-from cuenca_validations.types.requests import OperatorLoginRequest
-from pydantic import ConfigDict
+from cuenca_validations.types.requests import (
+    OperatorLoginRequest,
+    OperatorLoginUpdateRequest,
+)
+from pydantic import BaseModel, ConfigDict
 
 from ..http import Session, session as global_session
 from .base import Creatable
 
 
-class OperatorLogin(Creatable):
-    """Authenticate a company operator (portal personas morales).
+class OperatorLoginResponse(BaseModel):
+    """POST /operator-logins response (OTP challenge pending).
 
-    Authed endpoint: ``POST /operator-login``. Response ``id`` is the
-    Session.id; use it as ``X-Cuenca-SessionId`` on subsequent requests.
+    Requires cuenca-validations >= 2.1.48.
     """
 
-    _resource: ClassVar = 'operator-login'
+    id: str
+    operator_id: str
+    expires_at: dt.datetime
+    email_hint: str
 
-    id: Annotated[str, LogConfig(masked=True, unmasked_chars_length=4)]
+    model_config = ConfigDict(
+        json_schema_extra={
+            'example': {
+                'id': 'OLWqY5cvkISJOxHyEKjAKf8w',
+                'operator_id': 'OPWqY5cvkISJOxHyEKjAKf8w',
+                'expires_at': '2026-09-21T20:15:22Z',
+                'email_hint': 'ma****@aceros.com',
+            }
+        },
+    )
+
+
+class OperatorLoginSessionResponse(BaseModel):
+    """PATCH /operator-logins/{id} response (session after OTP).
+
+    Requires cuenca-validations >= 2.1.48.
+    """
+
+    id: str
     operator_id: str
     role: OperatorRole
     legal_person_id: str
@@ -30,6 +54,39 @@ class OperatorLogin(Creatable):
                 'role': 'authorizer',
                 'legal_person_id': 'USWqY5cvkISJOxHyEKjAKf8w',
             }
+        },
+    )
+
+
+class OperatorLogin(Creatable):
+    """Two-step operator portal login (Cuenca Empresas).
+
+    ``POST /operator-logins`` validates email/password and emails an OTP.
+    ``PATCH /operator-logins/{id}`` verifies the OTP and returns a Session.
+
+    Both routes are auth_exempt. Set ``X-Cuenca-SessionId`` only after
+    ``update`` / ``complete`` (response ``id`` is ``SE*``).
+    """
+
+    _resource: ClassVar = 'operator-logins'
+
+    id: Annotated[str, LogConfig(masked=True, unmasked_chars_length=4)]
+    operator_id: str
+    # Present after create (OTP challenge)
+    expires_at: Optional[dt.datetime] = None
+    email_hint: Optional[str] = None
+    # Present after update/complete (session)
+    role: Optional[OperatorRole] = None
+    legal_person_id: Optional[str] = None
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            'example': {
+                'id': 'OLWqY5cvkISJOxHyEKjAKf8w',
+                'operator_id': 'OPWqY5cvkISJOxHyEKjAKf8w',
+                'expires_at': '2026-09-21T20:15:22Z',
+                'email_hint': 'ma****@aceros.com',
+            }
         }
     )
 
@@ -42,10 +99,33 @@ def create(
         session: Session = global_session,
     ) -> 'OperatorLogin':
         req = OperatorLoginRequest(email=email, password=password)
-        login = cls._create(
+        return cls._create(
             session=session,
             email=str(req.email),
             password=req.password.get_secret_value(),
         )
+
+    @classmethod
+    def update(
+        cls,
+        login_id: str,
+        code: str,
+        *,
+        session: Session = global_session,
+    ) -> 'OperatorLogin':
+        req = OperatorLoginUpdateRequest(code=code)
+        resp = session.patch(f'/{cls._resource}/{login_id}', req.model_dump())
+        login = cls(**resp)
         session.headers['X-Cuenca-SessionId'] = login.id
         return login
+
+    @classmethod
+    def complete(
+        cls,
+        login_id: str,
+        code: str,
+        *,
+        session: Session = global_session,
+    ) -> 'OperatorLogin':
+        """Alias of ``update`` — verify OTP and create the Session."""
+        return cls.update(login_id, code, session=session)

cuenca/resources/operator_otps.py

@@ -0,0 +1,45 @@
+import datetime as dt
+from typing import Annotated, ClassVar
+
+from cuenca_validations.types import LogConfig
+from pydantic import ConfigDict
+
+from ..http import Session, session as global_session
+from .base import Creatable
+
+
+class OperatorOtp(Creatable):
+    """Email OTP challenge for authorized actions in operator portal.
+
+    Requires an active operator Session (from completed OperatorLogin).
+    ``POST /operator-otps`` creates an email OTP challenge and emails
+    a 6-digit code to the operator. Use the code with the ``X-Cuenca-OTP``
+    header in protected endpoints.
+
+    Expires in 5 minutes.
+    """
+
+    _resource: ClassVar = 'operator-otps'
+
+    id: Annotated[str, LogConfig(masked=True, unmasked_chars_length=4)]
+    expires_at: dt.datetime
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            'example': {
+                'id': 'OCWqY5cvkISJOxHyEKjAKf8w',
+                'expires_at': '2026-09-25T12:34:56Z',
+            }
+        }
+    )
+
+    @classmethod
+    def create(cls, *, session: Session = global_session) -> 'OperatorOtp':
+        """Create an email OTP challenge for the current operator Session.
+
+        Requires: X-Cuenca-SessionId header set (from OperatorLogin.update).
+        A 6-digit code is sent via email to the operator's registered address.
+
+        Returns: OperatorOtp with challenge id and expiration timestamp.
+        """
+        return cls._create(session=session)

cuenca/resources/operator_tokens.py

@@ -0,0 +1,61 @@
+import datetime as dt
+from typing import Annotated, ClassVar, Optional, Union
+
+from cuenca_validations.types import LogConfig, OperatorRole
+from pydantic import ConfigDict
+
+from ..http import Session, session as global_session
+from .base import Creatable
+
+
+class OperatorToken(Creatable):
+    """One-time invitation token to set an operator password.
+
+    ``POST /operator_tokens`` creates the invite in Authed and returns the
+    plaintext ``token`` once (for the magic-link email). Does not send email.
+    """
+
+    _resource: ClassVar = 'operator_tokens'
+
+    id: str
+    operator_id: str
+    email: str
+    legal_person_id: str
+    role: Union[OperatorRole, str]
+    token: Annotated[str, LogConfig(masked=True, unmasked_chars_length=4)]
+    created_at: Optional[dt.datetime] = None
+    expires_at: Optional[dt.datetime] = None
+    used_at: Optional[dt.datetime] = None
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            'example': {
+                'id': 'OTWqY5cvkISJOxHyEKjAKf8w',
+                'operator_id': 'OPWqY5cvkISJOxHyEKjAKf8w',
+                'email': 'maria.lopez@aceros.com',
+                'legal_person_id': 'USWqY5cvkISJOxHyEKjAKf8w',
+                'role': 'operator',
+                'token': 'plaintext-invite-secret',
+                'expires_at': '2026-09-24T21:30:56Z',
+            }
+        }
+    )
+
+    @classmethod
+    def create(
+        cls,
+        operator_id: str,
+        email: str,
+        legal_person_id: str,
+        role: Union[OperatorRole, str],
+        *,
+        session: Session = global_session,
+    ) -> 'OperatorToken':
+        role_value = role.value if isinstance(role, OperatorRole) else role
+        return cls._create(
+            session=session,
+            operator_id=operator_id,
+            email=email,
+            legal_person_id=legal_person_id,
+            role=role_value,
+        )

cuenca/version.py

@@ -1,3 +1,3 @@
-__version__ = '2.2.5'
+__version__ = '2.2.5.dev0'
 CLIENT_VERSION = __version__
 API_VERSION = '2020-03-19'

requirements.txt

@@ -1,3 +1,3 @@
 requests==2.32.3
-cuenca-validations==2.1.46
+cuenca-validations==2.1.48.dev2
 pydantic-extra-types==2.10.2

setup.py

@@ -24,7 +24,7 @@
     python_requires='>=3.9',
     install_requires=[
         'requests>=2.32.0',
-        'cuenca-validations>=2.1.46',
+        'cuenca-validations==2.1.48.dev2',
         'pydantic-extra-types>=2.10.0',
     ],
     classifiers=[

tests/resources/test_operator_logins.py

@@ -18,10 +18,10 @@ def session() -> Session:
 def test_operator_login_create(mock_request: MagicMock, session: Session):
     mock_request.return_value.ok = True
     mock_request.return_value.content = (
-        b'{"id":"SEWqY5cvkISJOxHyEKjAKf8w",'
+        b'{"id":"OLWqY5cvkISJOxHyEKjAKf8w",'
         b'"operator_id":"OPWqY5cvkISJOxHyEKjAKf8w",'
-        b'"role":"operator",'
-        b'"legal_person_id":"USWqY5cvkISJOxHyEKjAKf8w"}'
+        b'"expires_at":"2026-09-21T20:15:22",'
+        b'"email_hint":"ma****@aceros.com"}'
     )
 
     login = OperatorLogin.create(
@@ -30,18 +30,73 @@ def test_operator_login_create(mock_request: MagicMock, session: Session):
         session=session,
     )
 
-    assert login.id == 'SEWqY5cvkISJOxHyEKjAKf8w'
+    assert login.id == 'OLWqY5cvkISJOxHyEKjAKf8w'
     assert login.operator_id == 'OPWqY5cvkISJOxHyEKjAKf8w'
-    assert login.role == OperatorRole.operator
-    assert login.legal_person_id == 'USWqY5cvkISJOxHyEKjAKf8w'
-    assert session.headers['X-Cuenca-SessionId'] == login.id
-    assert 'X-Cuenca-LoginId' not in session.headers
+    assert login.email_hint == 'ma****@aceros.com'
+    assert login.expires_at is not None
+    assert login.role is None
+    assert login.legal_person_id is None
+    assert 'X-Cuenca-SessionId' not in session.headers
 
     mock_request.assert_called_once()
     _, kwargs = mock_request.call_args
     assert kwargs['method'] == 'post'
-    assert kwargs['url'] == 'https://sandbox.cuenca.com/operator-login'
+    assert kwargs['url'] == 'https://sandbox.cuenca.com/operator-logins'
     assert kwargs['json'] == {
         'email': 'maria@aceros.com',
         'password': 'supersecret',
     }
+
+
+@patch('cuenca.http.client.requests.Session.request')
+def test_operator_login_update(mock_request: MagicMock, session: Session):
+    mock_request.return_value.ok = True
+    mock_request.return_value.content = (
+        b'{"id":"SEWqY5cvkISJOxHyEKjAKf8w",'
+        b'"operator_id":"OPWqY5cvkISJOxHyEKjAKf8w",'
+        b'"role":"operator",'
+        b'"legal_person_id":"USWqY5cvkISJOxHyEKjAKf8w"}'
+    )
+
+    login = OperatorLogin.update(
+        'OLWqY5cvkISJOxHyEKjAKf8w',
+        '123456',
+        session=session,
+    )
+
+    assert login.id == 'SEWqY5cvkISJOxHyEKjAKf8w'
+    assert login.operator_id == 'OPWqY5cvkISJOxHyEKjAKf8w'
+    assert login.role == OperatorRole.operator
+    assert login.legal_person_id == 'USWqY5cvkISJOxHyEKjAKf8w'
+    assert session.headers['X-Cuenca-SessionId'] == login.id
+    assert 'X-Cuenca-LoginId' not in session.headers
+
+    _, kwargs = mock_request.call_args
+    assert kwargs['method'] == 'patch'
+    assert kwargs['url'] == (
+        'https://sandbox.cuenca.com/operator-logins/'
+        'OLWqY5cvkISJOxHyEKjAKf8w'
+    )
+    assert kwargs['json'] == {'code': '123456'}
+
+
+@patch('cuenca.http.client.requests.Session.request')
+def test_operator_login_complete_alias(
+    mock_request: MagicMock, session: Session
+):
+    mock_request.return_value.ok = True
+    mock_request.return_value.content = (
+        b'{"id":"SEWqY5cvkISJOxHyEKjAKf8w",'
+        b'"operator_id":"OPWqY5cvkISJOxHyEKjAKf8w",'
+        b'"role":"authorizer",'
+        b'"legal_person_id":"USWqY5cvkISJOxHyEKjAKf8w"}'
+    )
+
+    login = OperatorLogin.complete(
+        'OLWqY5cvkISJOxHyEKjAKf8w',
+        '654321',
+        session=session,
+    )
+
+    assert login.role == OperatorRole.authorizer
+    assert session.headers['X-Cuenca-SessionId'] == login.id