File: test_auto_close_of_data_objects__issue_456.py

package info (click to toggle)
python-irodsclient 3.0.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,352 kB
  • sloc: python: 16,650; xml: 525; sh: 104; awk: 5; sql: 3; makefile: 3
file content (54 lines) | stat: -rw-r--r-- 1,900 bytes parent folder | download
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
# This helper module can double as a Python script, allowing us to run the below
# test() method either within the current process or in a child process.  The
# method in question can thus be called by the following unit tests so that we may assert
# proper data object auto-closing functionality under these respective scenarios:
#
#    irods.test.data_obj_test.TestDataObjOps.test_data_objects_auto_close_on_function_exit__issue_456
#    irods.test.data_obj_test.TestDataObjOps.test_data_objects_auto_close_on_process_exit__issue_456

import contextlib

try:
    import irods.client_configuration as config
except ImportError:
    pass
from datetime import datetime
import os
from irods.test import helpers


@contextlib.contextmanager
def auto_close_data_objects(value):
    if "config" not in globals():
        yield
        return
    ORIGINAL_VALUE = config.data_objects.auto_close
    try:
        config.data_objects.auto_close = value
        yield
    finally:
        config.data_objects.auto_close = ORIGINAL_VALUE


def test(return_locals=True):
    with auto_close_data_objects(True):
        expected_content = "content"
        ses = helpers.make_session()
        name = "/{0.zone}/home/{0.username}/{1}-object.dat".format(
            ses, helpers.unique_name(os.getpid(), datetime.now())
        )
        f = ses.data_objects.open(name, "w")
        f.write(expected_content.encode("utf8"))
        L = locals()
        # By default, ses and f will be automatically exported to calling frame (with L being returned),
        # but by specifying a list/tuple of keys we can export only those specific locals by name.
        return (
            L
            if not isinstance(return_locals, (tuple, list))
            else [L[k] for k in return_locals]
        )


if __name__ == "__main__":
    test_output = test()
    print("{name} {expected_content}".format(**test_output))