invenio-requests/tests/services/requests/test_requests_tasks.py at b29e7d40ff1e5bef4b7cf835f2e7eb0a55ae44d3 · inveniosoftware/invenio-requests · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
# -*- coding: utf-8 -*-
#
# Copyright (C) 2022-2026 Graz University of Technology.
#
# Invenio-Requests is free software; you can redistribute it and/or
# modify it under the terms of the MIT License; see LICENSE file for more
# details.

"""Tasks tests."""

from datetime import datetime, timedelta, timezone

from invenio_access.permissions import system_identity
from invenio_search.engine import dsl

from invenio_requests.records.api import Request
from invenio_requests.tasks import check_expired_requests


def test_check_expired_requests(
    app, identity_simple, create_request, submit_request, requests_service
):
    """Test if the expired system field works as intended."""
    now = datetime.now(timezone.utc).replace(tzinfo=None)

    # created only should not be picked up
    created_request = create_request(identity=identity_simple, expires_at=now)
    Request.index.refresh()
    check_expired_requests()
    Request.index.refresh()
    request_list = requests_service.search(
        identity=system_identity,
        extra_filter=dsl.query.Bool(
            "must",
            must=[
                dsl.Q("term", **{"is_closed": True}),
            ],
        ),
    )
    assert request_list.total == 0

    # no expires_at should not be touched
    s1 = submit_request(identity_simple)
    Request.index.refresh()
    check_expired_requests()
    Request.index.refresh()
    request_list = requests_service.search(
        identity=system_identity,
        extra_filter=dsl.query.Bool(
            "must",
            must=[
                dsl.Q("term", **{"is_closed": True}),
            ],
        ),
    )
    assert request_list.total == 0

    # expiry date in future should not be touched
    s2 = submit_request(identity_simple, expires_at=now + timedelta(days=1))
    Request.index.refresh()
    check_expired_requests()
    Request.index.refresh()
    request_list = requests_service.search(
        identity=system_identity,
        extra_filter=dsl.query.Bool(
            "must",
            must=[
                dsl.Q("term", **{"is_closed": True}),
            ],
        ),
    )
    assert request_list.total == 0

    s3 = submit_request(identity_simple, expires_at=now)
    Request.index.refresh()
    check_expired_requests()
    Request.index.refresh()
    request_list = requests_service.search(
        identity=system_identity,
        extra_filter=dsl.query.Bool(
            "must",
            must=[
                dsl.Q("term", **{"is_closed": True}),
            ],
        ),
    )
    assert request_list.total == 1