sip2: implements selfcheck renewal

* Allows the patron to do a renewal on the selfcheck machine.
* Adds CLI for selfcheck terminal creation.
* Adds renew handler.

Lauren-D <[email protected]>

Author: lauren-d

rero_ils/config.py

@@ -2708,6 +2708,7 @@ def _(x):
         circulation_handlers=dict(
             checkout='rero_ils.modules.selfcheck.api:selfcheck_checkout',
             checkin='rero_ils.modules.selfcheck.api:selfcheck_checkin',
+            renew='rero_ils.modules.selfcheck.api:selfcheck_renew',
         )
     )
 )

rero_ils/modules/cli.py

@@ -75,6 +75,7 @@
 from .monitoring import Monitoring
 from .operation_logs.cli import create_operation_logs, dump_operation_logs
 from .patrons.cli import import_users, users_validate
+from .selfcheck.cli import terminal_create
 from .tasks import process_bulk_queue
 from .utils import bulk_load_metadata, bulk_load_pids, bulk_load_pidstore, \
     bulk_save_metadata, bulk_save_pids, bulk_save_pidstore, \
@@ -132,6 +133,7 @@ def utils():
 
 
 utils.add_command(users_validate)
+utils.add_command(terminal_create)
 
 
 def queue_count():

rero_ils/modules/selfcheck/api.py

@@ -28,6 +28,7 @@
     map_item_circulation_status, map_media_type
 from ..documents.api import Document
 from ..documents.utils import title_format_text_head
+from ..errors import NoCirculationAction
 from ..items.api import Item
 from ..items.models import ItemNoteTypes
 from ..libraries.api import Library
@@ -351,9 +352,8 @@ def selfcheck_checkout(transaction_user_pid, item_barcode, patron_barcode,
                         item_pid=item.pid,
                         selfcheck_terminal_id=str(terminal.id),
                     )
-                    loan_pid = data[LoanAction.CHECKOUT].get('pid')
-                    loan = Loan.get_record_by_pid(loan_pid)
-                    if loan:
+                    if data[LoanAction.CHECKOUT]:
+                        loan = data[LoanAction.CHECKOUT]
                         checkout['checkout'] = True
                         checkout['due_date'] = loan.get_loan_end_date(
                             time_format=None, language=language)
@@ -413,6 +413,7 @@ def selfcheck_checkin(transaction_user_pid, item_barcode, **kwargs):
                         transaction_user_pid=staffer.pid,
                         transaction_library_pid=terminal.library_pid,
                         item_pid=item.pid,
+                        selfcheck_terminal_id=str(terminal.id),
                     )
                     if data[LoanAction.CHECKIN]:
                         checkin['checkin'] = True
@@ -428,3 +429,71 @@ def selfcheck_checkin(transaction_user_pid, item_barcode, **kwargs):
                 _('Error encountered: please contact a librarian'))
             raise SelfcheckCirculationError('self checkin failed', checkin)
         return checkin
+
+
+def selfcheck_renew(transaction_user_pid, item_barcode, **kwargs):
+    """SIP2 handler to perform renew.
+
+    Perform renew action received from the selfcheck.
+    :param transaction_user_pid: identifier of the staff user.
+    :param item_barcode: item identifier.
+    :return: The SelfcheckRenew object.
+    """
+    if check_sip2_module():
+        from invenio_sip2.errors import SelfcheckCirculationError
+        from invenio_sip2.models import SelfcheckFeeType, SelfcheckRenew
+
+        terminal = SelfcheckTerminal.find_terminal(
+            name=kwargs.get('terminal'))
+        item = Item.get_item_by_barcode(
+            barcode=item_barcode,
+            organisation_pid=terminal.organisation_pid
+        )
+        document = Document.get_record_by_pid(item.document_pid)
+
+        renew = SelfcheckRenew(
+            title_id=title_format_text_head(document.get('title'))
+        )
+        with current_app.test_request_context() as ctx:
+            language = kwargs.get('language', current_app.config
+                                  .get('BABEL_DEFAULT_LANGUAGE'))
+            ctx.babel_locale = language
+            try:
+                staffer = Patron.get_record_by_pid(transaction_user_pid)
+                if staffer.is_librarian:
+                    # get renewal count
+                    renewal_count = item.get_extension_count()
+                    if renewal_count > 1:
+                        renew['renewal'] = True
+                    # do extend loan
+                    result, data = item.extend_loan(
+                        transaction_user_pid=staffer.pid,
+                        transaction_library_pid=terminal.library_pid,
+                        item_pid=item.pid,
+                        selfcheck_terminal_id=str(terminal.id),
+                    )
+                    if data[LoanAction.EXTEND]:
+                        loan = data[LoanAction.EXTEND]
+                        renew['success'] = True
+                        renew['due_date'] = loan.get_loan_end_date(
+                            time_format=None, language=language)
+                        transaction = PatronTransaction. \
+                            get_last_transaction_by_loan_pid(
+                                loan_pid=loan.pid,
+                                status='open')
+                        if transaction:
+                            # TODO: map transaction type
+                            renew['fee_type'] = SelfcheckFeeType.OVERDUE
+                            renew['fee_amount'] = transaction.total_amount
+                            renew['currency_type'] = transaction.currency
+                        # TODO: When is possible, try to return fields:
+                        #       magnetic_media, resensitize
+
+            except NoCirculationAction:
+                renew.get('screen_messages', []).append(
+                    _('No circulation action is possible'))
+            except Exception:
+                renew.get('screen_messages', []).append(
+                    _('Error encountered: please contact a librarian'))
+                raise SelfcheckCirculationError('self renewal failed', renew)
+            return renew

rero_ils/modules/selfcheck/cli.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+#
+# RERO ILS
+# Copyright (C) 2021 RERO
+# Copyright (C) 2021 UCLouvain
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""Click command-line utilities."""
+
+import click
+from flask.cli import with_appcontext
+from invenio_db import db
+from invenio_oauth2server.cli import process_scopes, process_user
+
+from rero_ils.modules.locations.api import search_location_by_pid
+
+from .models import SelfcheckTerminal
+
+
+@click.command('terminal_create')
+@click.option('-n', '--name', required=True)
+@click.option(
+    '-u', '--user', required=True, callback=process_user,
+    help='User ID or email.')
+@click.option(
+    '-l', '--location-pid', required=True)
+@click.option(
+    '-s', '--scope', 'scopes', multiple=True, callback=process_scopes)
+@click.option('-i', '--internal', is_flag=True)
+@click.option(
+    '-t', '--access_token', 'access_token', required=False,
+    help='personalized access_token.')
+@with_appcontext
+def terminal_create(name, user, location_pid, scopes, internal,
+                    access_token):
+    """Create a personal OAuth token."""
+    from rero_ils.modules.cli import create_personal  # avoid circular import
+
+    location = search_location_by_pid(location_pid)
+    if location:
+        token = create_personal(
+            name, user.id, scopes=scopes, is_internal=internal,
+            access_token=access_token)
+        selfcheck_terminal = SelfcheckTerminal(
+            name=name,
+            access_token=token.access_token,
+            organisation_pid=location.organisation['pid'],
+            library_pid=location.library['pid'],
+            location_pid=location_pid
+        )
+        db.session.add(selfcheck_terminal)
+        db.session.commit()
+        click.secho(f'login: {name}', fg='blue')
+        click.secho(f'token: {token.access_token}', fg='blue')

tests/api/selfcheck/test_selfcheck.py

@@ -33,7 +33,7 @@
 from rero_ils.modules.notifications.dispatcher import Dispatcher
 from rero_ils.modules.selfcheck.api import authorize_patron, enable_patron, \
     item_information, patron_information, selfcheck_checkin, \
-    selfcheck_checkout, selfcheck_login, system_status, \
+    selfcheck_checkout, selfcheck_login, selfcheck_renew, system_status, \
     validate_patron_account
 from rero_ils.modules.selfcheck.utils import check_sip2_module
 
@@ -259,11 +259,11 @@ def test_item_information(client, librarian_martigny,
     assert res.status_code == 200
 
 
-def test_selfcheck_checkout(client, selfcheck_librarian_martigny,
-                            selfcheck_patron_martigny, loc_public_martigny,
-                            item_lib_martigny, librarian_martigny,
-                            librarian2_martigny,  circulation_policies):
-    """Test selfcheck checkout."""
+def test_selfcheck_circulation(client, selfcheck_librarian_martigny, document,
+                               librarian_martigny, librarian2_martigny,
+                               loc_public_martigny, selfcheck_patron_martigny,
+                               item_lib_martigny, circulation_policies):
+    """Test selfcheck circulation operation."""
     patron_barcode = selfcheck_patron_martigny \
         .get('patron', {}).get('barcode')[0]
     item_barcode = item_lib_martigny.get('barcode')
@@ -278,49 +278,15 @@ def test_selfcheck_checkout(client, selfcheck_librarian_martigny,
     assert checkout.is_success
     assert checkout.due_date
 
-    # librarian checkin
-    login_user_via_session(client, librarian2_martigny.user)
-    res, _ = postdata(
-        client,
-        'api_item.checkin',
-        dict(
-            item_pid=item_lib_martigny.pid,
-            transaction_user_pid=librarian2_martigny.pid,
-            transaction_location_pid=loc_public_martigny.pid
-        )
+    # selfcheck renew
+    renew = selfcheck_renew(
+        transaction_user_pid=librarian_martigny.pid,
+        item_barcode=item_barcode, patron_barcode=patron_barcode,
+        terminal=selfcheck_librarian_martigny.name
     )
-    assert res.status_code == 200
-
-
-def test_selfcheck_checkin(client, selfcheck_librarian_martigny,
-                           librarian_martigny, librarian2_martigny,
-                           loc_public_martigny, selfcheck_patron_martigny,
-                           item_lib_martigny, document, circulation_policies):
-    """Test selfcheck checkin."""
-    patron_barcode = selfcheck_patron_martigny \
-        .get('patron', {}).get('barcode')[0]
-    item_barcode = item_lib_martigny.get('barcode')
-
-    # librarian checkout
-    login_user_via_session(client, librarian2_martigny.user)
-    res, data = postdata(client, 'api_item.checkout', dict(
-        item_pid=item_lib_martigny.pid,
-        patron_pid=selfcheck_patron_martigny.pid,
-        transaction_location_pid=loc_public_martigny.pid,
-        transaction_user_pid=librarian2_martigny.pid
-    ))
-    assert res.status_code == 200
-
-    # test selfcheck checkin with invalid item barcode
-    with pytest.raises(Exception):
-        checkin = selfcheck_checkin(
-            transaction_user_pid=librarian_martigny.pid,
-            patron_barcode=patron_barcode,
-            item_barcode='wrong_item_barcode',
-            terminal=selfcheck_librarian_martigny.name
-        )
-        assert checkin
-        assert not checkin.is_success
+    assert renew
+    assert renew.is_success
+    assert renew.due_date
 
     # selfcheck checkin
     checkin = selfcheck_checkin(