Skip to content

fix 63714 #1953

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Ooka-Masashi merged 1 commit into from
Oct 29, 2026
Merged

fix 63714 #1953

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions modules/weko-deposit/tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
from invenio_files_rest.views import blueprint as invenio_files_rest_blueprint
from invenio_files_rest.models import Bucket, Location, ObjectVersion
from invenio_pidstore.models import PersistentIdentifier, PIDStatus, Redirect
from invenio_records.models import RecordMetadata
from invenio_i18n import InvenioI18N
from weko_index_tree.api import Indexes
from weko_index_tree.models import Index
Expand Down Expand Up @@ -301,6 +302,41 @@ def db(app):
# drop_database(str(db_.engine.url))


@pytest.fixture()
def item_edit_permission_records(db):
"""Create records and recids for item edit permission tests."""
record_uuid = uuid.uuid4()
record = RecordMetadata(id=record_uuid, json={}, version_id=1)
db.session.add(record)
record_pid = PersistentIdentifier.create(
"recid",
"item-edit",
object_type="rec",
object_uuid=record_uuid,
status=PIDStatus.REGISTERED,
)
parent_pid = PersistentIdentifier.create(
"recid",
"item-edit-parent",
object_type="rec",
object_uuid=record_uuid,
status=PIDStatus.REGISTERED,
)
record_missing_pid = PersistentIdentifier.create(
"recid",
"item-edit-orphan",
object_type="rec",
object_uuid=uuid.uuid4(),
status=PIDStatus.REGISTERED,
)
db.session.commit()
return {
"record_pid": record_pid.pid_value,
"parent_pid": parent_pid.pid_value,
"record_missing_pid": record_missing_pid.pid_value,
}


@pytest.fixture()
def records(db):
record_data = json_data("data/test_records.json")
Expand Down
149 changes: 148 additions & 1 deletion modules/weko-deposit/tests/test_rest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,160 @@

from elasticsearch import ElasticsearchException
from sqlalchemy.exc import SQLAlchemyError
from flask import url_for
from flask import session, url_for
from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
from weko_deposit.api import WekoDeposit, WekoRecord
from weko_deposit.rest import require_item_edit_permission
from invenio_accounts.testutils import login_user_via_session


# .tox/c1/bin/pytest --cov=weko_deposit tests/test_rest.py -vv -s --cov-branch --cov-report=term --cov-report=html --basetemp=/code/modules/weko-deposit/.tox/c1/tmp


# .tox/c1/bin/pytest --cov=weko_deposit tests/test_rest.py::test_require_item_edit_permission -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-deposit/.tox/c1/tmp
def test_require_item_edit_permission(app, item_edit_permission_records):
def create_protected_view():
view = MagicMock(return_value="permitted")
return view, require_item_edit_permission(view)

def permission_factory(can_edit):
permission = MagicMock()
permission.can.return_value = can_edit
return patch(
"weko_items_ui.permissions.edit_permission_factory",
return_value=permission,
)

# 未認証かつゲストトークンなしの場合は401を返す。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=False),
), app.test_request_context():
with pytest.raises(Unauthorized) as exc_info:
protected_view(pid_value=item_edit_permission_records["record_pid"])

assert exc_info.value.code == 401
view.assert_not_called()

# 認証済みかつゲストトークンありの場合は401を返さない。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=True),
), permission_factory(False), app.test_request_context():
session["guest_token"] = "guest-token"
assert protected_view(
pid_value=item_edit_permission_records["record_pid"]
) == "permitted"

view.assert_called_once_with(
pid_value=item_edit_permission_records["record_pid"]
)

# 未認証でもゲストトークンがあれば401を返さない。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=False),
), permission_factory(False), app.test_request_context():
session["guest_token"] = "guest-token"
assert protected_view(
pid_value=item_edit_permission_records["record_pid"]
) == "permitted"

view.assert_called_once_with(
pid_value=item_edit_permission_records["record_pid"]
)

# バージョン付きPIDは親のrecidを使って編集権限を確認する。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=True),
), permission_factory(True), app.test_request_context():
versioned_pid = "{}.1".format(
item_edit_permission_records["parent_pid"]
)
assert protected_view(pid_value=versioned_pid) == "permitted"

view.assert_called_once_with(pid_value=versioned_pid)

# PIDが存在しない場合は404とItem not foundを返す。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=True),
), app.test_request_context():
with pytest.raises(NotFound) as exc_info:
protected_view(pid_value="item-edit-missing")

assert exc_info.value.code == 404
assert exc_info.value.description == "Item not found"
view.assert_not_called()

# PIDに対応するレコードが存在しない場合は404とItem not foundを返す。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=True),
), app.test_request_context():
with pytest.raises(NotFound) as exc_info:
protected_view(
pid_value=item_edit_permission_records["record_missing_pid"]
)

assert exc_info.value.code == 404
assert exc_info.value.description == "Item not found"
view.assert_not_called()

# 編集権限もゲストトークンもない場合は403を返す。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=True),
), permission_factory(False), app.test_request_context():
with pytest.raises(Forbidden) as exc_info:
protected_view(
pid_value=item_edit_permission_records["record_pid"]
)

assert exc_info.value.code == 403
assert exc_info.value.description == (
"You do not have permission to edit this item"
)
view.assert_not_called()

# 編集権限があればゲストトークンがなくても403を返さない。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=True),
), permission_factory(True), app.test_request_context():
assert protected_view(
pid_value=item_edit_permission_records["record_pid"]
) == "permitted"

view.assert_called_once_with(
pid_value=item_edit_permission_records["record_pid"]
)

# 編集権限がなくてもゲストトークンがあれば403を返さない。
view, protected_view = create_protected_view()
with patch(
"weko_deposit.rest.current_user",
MagicMock(is_authenticated=False),
), permission_factory(False), app.test_request_context():
session["guest_token"] = "guest-token"
assert protected_view(
pid_value=item_edit_permission_records["record_pid"]
) == "permitted"

view.assert_called_once_with(
pid_value=item_edit_permission_records["record_pid"]
)


# def test_publish(app, location):
# deposit = WekoDeposit.create({})
# kwargs = {
Expand Down
6 changes: 4 additions & 2 deletions modules/weko-deposit/weko_deposit/rest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import sys
from functools import wraps
from wsgiref.util import request_uri
from flask import session

@sourcery-ai sourcery-ai Bot Oct 28, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚪ Low · Import ordering check fails

from flask import session is separate from the existing Flask import and appears before imports that should precede Flask, so isort . rewrites this block and the formatting check reports a finding.

Merge session into the existing Flask import and order the imports with isort.

Prompt for AI agents 
In `modules/weko-deposit/weko_deposit/rest.py` at line 27:

**Import ordering check fails**

`from flask import session` is separate from the existing Flask import and appears before imports that should precede Flask, so `isort .` rewrites this block and the formatting check reports a finding.

Merge `session` into the existing Flask import and order the imports with isort.

From AGENTS.md L70-73:

- フォーマッター: **Black** を使用（`black .` でソースコードを整形）
- リンター: **Flake8** を使用（`flake8` で静的解析チェック）
- インポート順の整理: **isort** を使用（`isort .` でインポート並び替え）
- これらのフォーマットチェックはコミット前に必ず実行し、指摘がない状態にしてください


import redis
from redis import sentinel
Expand Down Expand Up @@ -69,7 +70,8 @@ def require_item_edit_permission(f):
def decorated(*args, **kwargs):
# 循環 import を避けるため関数内で読み込む
from weko_items_ui.permissions import edit_permission_factory
if not (current_user and current_user.is_authenticated):
guest_token = session.get("guest_token")
if not (current_user and current_user.is_authenticated) and not guest_token:
abort(401)
pid_value = kwargs.get('pid_value')
pid_value = str(getattr(pid_value, 'value', pid_value))
Expand All @@ -86,7 +88,7 @@ def decorated(*args, **kwargs):
r = RecordMetadata.query.filter_by(id=pid.object_uuid).first()
if r is None:
abort(404, "Item not found")
if not edit_permission_factory(r.json).can():
if not edit_permission_factory(r.json).can() and not guest_token:
abort(403, "You do not have permission to edit this item")
return f(*args, **kwargs)
return decorated
Expand Down
Loading
Loading